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 | |||||||||||||||
|
|
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 | |||||||||||||||
|
|
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 |
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 |