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 | |
class DemoStage(BaseStage): pass | DemoStage( enable=True, name='demo', save_to_intermediate=False, expand_result=False, ui_only=None, halt_on_exception=False ) |
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.
In the following example the default values where omitted, leaving only the new and required values
Class | Object | |
class DemoStage(BaseStage): message: str parameter_a: int parameter_b: float | 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 | |
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') | 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)
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.
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:
self
intermediate
final
request
Request allows access to the user information currently sending the request
user: User = request.state.user