Create the stage file
A descriptive name works the best
on the stage file
from typing import Any, Dict from starlette.requests import Request from app.pipeline import BaseStage, BaseStageImpl from utils.data import ImmutableDict
class DemoStage(BaseStage): pass
Here is where the logic should be
class DemoStageImpl(BaseStageImpl[DemoStage]):
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
from .demo import DemoStage, DemoStageImpl
on the __init__ file of the stages package
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 *