Create the stage file
Tip |
---|
A descriptive name works the best |
on the stage file
Code Block | ||||
---|---|---|---|---|
| ||||
from typing import Any, Dict
from starlette.requests import Request
from app.pipeline import BaseStage, BaseStageImpl
from utils.data import ImmutableDict |
Code Block | ||||
---|---|---|---|---|
| ||||
class DemoStage(BaseStage):
pass |
Here is where the logic should be
Code Block | ||||
---|---|---|---|---|
| ||||
class DemoStageImpl(BaseStageImpl[DemoStage]): |
Code Block | ||||
---|---|---|---|---|
| ||||
class DemoStageImpl(BaseStageImpl[DemoStage]):
async def process(self, intermediate: Dict[str, Any], final: ImmutableDict[str, Any], request: Request) -> Any:
# your logic in here
pass
async def post_process(self, intermediate: Dict[str, Any], final: ImmutableDict[str, Any], request: Request) -> Any:
# your logic in here
pass
async def get_ui_config(self, request: Request) -> Any:
# your logic in here
pass |
on the __init__ file
Code Block | ||||
---|---|---|---|---|
| ||||
from .demo import DemoStage, DemoStageImpl |
on the __init__ file of the stages package
Code Block | ||||
---|---|---|---|---|
| ||||
from .dynamic_agg import *
from .dynamic_results import *
from .filter import *
from .highlight import *
from .query import *
from .saga_query import *
from .sample_query import *
from .parallel_wrapper import *
# Your new stage
from .demo import * |