You are viewing an old version of this page. View the current version.

Compare with Current View Page History

« Previous Version 3 Next »

Before Start

Building a Stage


1) Create the package

  • Start by creating a python file for the pipeline configuration in the folder app/pipeline/stages
  • Create the __init__ file

  • Create the stage file

    A descriptive name works the best



on the stage file


1) Add the imports

  • Special attention to BaseStage & BaseStageImpl
from typing import Any, Dict

from starlette.requests import Request

from app.pipeline import BaseStage, BaseStageImpl
from utils.data import ImmutableDict

2) Declare the Stage class inheriting from BaseStage

  • Essentially a pydantic model, the is where your configuration will be
class DemoStage(BaseStage):
    pass

3) Declare the Stage Implementation 

  • Inheriting BaseStageImpl, with the Stage class (DemoStage) as a parameter

  • Here is where the logic should be

class DemoStageImpl(BaseStageImpl[DemoStage]):

4) Add the abstract methods and __init__ if required

  • These are the only required conditions for this class, everything else depends on your implementation
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


5) Add the import for you stage:

  • This is for a better access when importing the stage
from .demo import DemoStage, DemoStageImpl



on the __init__ file of the stages package


6) Add the import all from the package of your stage

  • This is for a better access when importing the stage,
  • At this point you can import the stage with app.pipeline.stages
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 *
  • No labels