Giuliano Mega
11/10/2022, 11:54 AMclass FlowParameters(BaseModel):
requests_per_minute: int = 10
ids: Optional[List[str]] = None
@flow(
description='Imports and enriches Entities.',
task_runner=SequentialTaskRunner()
)
def token_pipeline(parameters: FlowParameters):
new_entities = fetch_new_entities(parameters.ids)
enrich_entity.map(new_entities, parameters.requests_per_minute)
If I call my flow from Prefect Cloud and pass it some parameters (run Custom), all runs smoothly. If, on the other hand, I try to run it with the default parameters, I get:
Validation of flow parameters failed with error:
Traceback (most recent call last):
File "/usr/local/lib/python3.10/site-packages/prefect/engine.py", line 275, in retrieve_flow_then_begin_flow_run
parameters = flow.validate_parameters(flow_run.parameters)
File "/usr/local/lib/python3.10/site-packages/prefect/flows.py", line 324, in validate_parameters
raise ParameterTypeError.from_validation_error(exc) from None
prefect.exceptions.ParameterTypeError: Flow run received invalid parameters:
- parameters: field required
Which is weird. Expected behavior would be having the flow run with default parameters. Can't seem to understand what I'm doing wrong from reading the docs, any ideas? 🙂Ryan Peden
11/10/2022, 1:20 PMdef token_pipeline(parameters: FlowParameters = FlowParameters())
Giuliano Mega
11/10/2022, 2:41 PMGiuliano Mega
11/10/2022, 2:42 PM