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

Compare with Current View Page History

« Previous Version 11 Next »

With sapi.py (project manager)

Available since version 0.5

On the root of the GAIA API project execute on the terminal the following command:

python sapi.py new stage <name>

This will generate a bare minimum stage with all the steps and changes view on the manual section

Manual

Use this section if sapi.py is not available in your distribution

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 (Configuration)

  • Essentially a pydantic model, this is where your configuration will be
  • This is the Stage configuration set on the pipeline
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
  1. __init__

    1. Stage Initialization, one time code execution

  2. process

    1. Main stage body

    2. Accepts 3 parameters (intermediate, final, request)

    3. If returns a value, value will end in the final response

    4. If no return no value added to the response

  3. post_process

    1. Alternative or secondary method, if need

    2. Execute after all the stages have executed process

    3. Accepts 3 parameters (intermediate, final, request)

    4. If returns a value, value will end in the final response

    5. If no return no value added to the response

  4. get_ui_config

    1. For user interface only

    2. Accepts 1 parameter (request)

    3. Return UI config stored in the ui_only parameter

class DemoStageImpl(BaseStageImpl[DemoStage]):

    def __init__(self, config: DemoStage):
        super().__init__(config)
    
    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 *

Bare Minimum Stage

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


class DemoStageImpl(BaseStageImpl[DemoStage]):

    def __init__(self, config: DemoStage):
        super().__init__(config)

    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
  • No labels