With gapi.py (project manager)

Available since version 0.5

Rename in version 3.0 from sapi to gapi

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

python gapi.py new stage <name>

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

Documentation below does not apply for version >= 3.0.0

Manual

Use this section if gapi.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