Use this section if sapi.py is not available in your distribution
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]): |
Stage Initialization, one time code execution
Main stage body
Accepts 3 parameters (intermediate, final, request)
If returns a value, value will end in the final response
If no return no value added to the response
Alternative or secondary method, if need
Execute after all the stages have executed process
Accepts 3 parameters (intermediate, final, request)
If returns a value, value will end in the final response
If no return no value added to the response
For user interface only
Accepts 1 parameter (request)
Return UI config stored in the ui_only parameter
Code Block | ||||
---|---|---|---|---|
| ||||
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
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 * |
Code Block | ||||
---|---|---|---|---|
| ||||
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 |