mondras
01/08/2025, 6:31 PMMarvin
01/08/2025, 6:31 PMpython
from prefect import flow
@flow
def my_flow(param1: int, param2: str = "default"):
print(f"Parameter 1: {param1}")
print(f"Parameter 2: {param2}")
# Running the flow with parameters
my_flow(42, param2="Hello, Prefect!")
### Key Points:
- Parameters: You can define flow parameters using function arguments. Both positional and keyword arguments are supported, but when using the Prefect API, you must use keyword arguments to override defaults.
- Validation: Prefect validates parameters before running a flow. If invalid parameters are provided, the flow will not transition to a running state.
- Size Limit: Be aware that flow run parameters cannot exceed 512kb in size.
For more details, you can refer to the Prefect documentation on writing flows.
If you have any specific constraints or needs related to your Python/Rust integration with PyO3, feel free to ask!mondras
01/08/2025, 6:32 PMMarvin
01/08/2025, 6:32 PMpython
from prefect import flow
from datetime import datetime
from pydantic import BaseModel
class MyParams(BaseModel):
start_date: datetime
threshold: float = 0.5
@flow
def my_flow(params: MyParams):
print(f"Start Date: {params.start_date}")
print(f"Threshold: {params.threshold}")
# Example usage
my_flow(params=MyParams(start_date=datetime.now()))
### Additional Resources:
- Prefect Documentation on Writing Flows
This approach ensures that when you or others are submitting flows via the UI, the parameters are well-understood and correctly utilized. If you need further customization or have specific questions about integrating this with your Python/Rust stack, feel free to ask!mondras
01/08/2025, 6:34 PMMarvin
01/08/2025, 6:34 PMmondras
01/08/2025, 6:34 PMMarvin
01/08/2025, 6:34 PMmondras
01/08/2025, 6:38 PMMarvin
01/08/2025, 6:38 PMNate
01/08/2025, 6:43 PMNate
01/08/2025, 6:43 PMNate
01/08/2025, 6:43 PMMarvin
01/08/2025, 6:43 PMNate
01/08/2025, 6:43 PMmondras
01/08/2025, 6:51 PMNate
01/08/2025, 6:57 PMfrom typing import Annotated
from pydantic import Field
from prefect import flow
SpecialString = Annotated[
str, Field(json_schema_extra={"title": "something informative"})
]
@flow
def my_flow(special_string: SpecialString) -> None:
print(special_string)
if __name__ == "__main__":
my_flow.serve()
Nate
01/08/2025, 6:57 PMNate
01/08/2025, 6:58 PMAnnotated
should give you the ability to add info to builtin typesmondras
01/08/2025, 6:58 PM