As in all other configurations the stage configuration is a Pydantic model, so all advantages of Pydantic applied to the stage configuration (e.g, validations data transformation)
The BaseStage class is a Pydantic model with general configuration setup, were all base parameters are optional except for name
Class
Object
Code Block | ||||
---|---|---|---|---|
| ||||
class DemoStage(BaseStage):
pass |
Code Block | ||||
---|---|---|---|---|
| ||||
DemoStage(
enable=True,
name='demo',
save_to_intermediate=False,
expand_result=False,
ui_only=None,
halt_on_exception=False
) |
Table of Contents |
---|
To add parameters, just class variables and their type. In the example below, all parameters are required, the only validation besides required, is their type.
Info |
---|
In the following example the default values where omitted, leaving only the new and required values |
Class
Object
Code Block | ||||
---|---|---|---|---|
| ||||
class DemoStage(BaseStage):
message: str
parameter_a: int
parameter_b: float |
Code Block | ||||
---|---|---|---|---|
| ||||
DemoStage(
name='demo',
message='Add this to response',
parameter_a=5,
parameter_a=3.6
) |
All examples above are Pydantic models, but none is taking advantage of it, so if we add this feature we would get something like this
Class
Object
Code Block | ||||
---|---|---|---|---|
| ||||
from typing import Optional
from pydantic import Field
class DemoStage(BaseStage):
message: Optional[str] = Field(title='Title Message',
description='Message to put on the response body')
parameter_a: int = Field(default=1, ge=0, lt=10,
description='Parameter to do something')
parameter_b: float = Field(..., default=0.5, ge=0, lt=10,
description='Parameter to do something else') |
Code Block | ||||
---|---|---|---|---|
| ||||
DemoStage(
name='demo',
message='Add this to response',
parameter_a=5,
parameter_a=3.6
) |
Here we are:
Adding restrictions and verification to parameter_a and parameter_b with ge (greater or equal) and lt (lower than)
Info |
---|
If the value set does not met the gt and lt parameter requirements, an exception is thrown |
Making parameter_b required even if we set a default, meaning only by setting parameter_b as None, will the default by use
Initialization of the stage (constructor), executed only once when the pipeline is build, or when the pipeline is updated.
The init of the stage must always accept a config parameter, with the type of the Stage class.
Code Block | ||||
---|---|---|---|---|
| ||||
def __init__(self, config: DemoStage):
super().__init__(config)
import requests
url = "https://api.example.com/data" # Replace with your desired URL
response = requests.get(url)
if response.status_code == 200:
print(f"Response: {response.text}")
else:
print(f"GET request failed with status code: {response.status_code}") |
Useful for:
Each stage has 3 required methods, process being the main one, post_process the alternative or secondary executed only after process and get_ui_config currently under development and still not a definitive method
This methods share the same parameters:
Column | ||
---|---|---|
| ||
self
|
Column | ||
---|---|---|
| ||
intermediate
|
Column | ||
---|---|---|
| ||
final
|
Column | |||||||||
---|---|---|---|---|---|---|---|---|---|
| |||||||||
request
|
Create the stage file
Tip |
---|
A descriptive name works the best |
Table of Contents |
---|
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 |