Arthur Rey
03/06/2024, 5:48 AMMarvin
03/06/2024, 5:48 AMparameter_openapi_schema
argument for a Prefect deployment function:
python
import time
from prefect import flow, task
from pydantic import BaseModel, Field
class Data(BaseModel):
key: str
value: dict = Field(default_factory=dict)
@task
def fetch_data_for_key(key: str) -> Data:
"""A task that simulates fetching data from some source"""
return Data(key=key, value={"data": "some data"})
@task
def process_data(data: Data) -> Data:
"""A task that simulates processing data"""
print(f"Processing {data!r} (I swear)")
time.sleep(3)
return data
@flow
def work_i_need_to_do_sometimes(keys: list[str]):
"""Work you might need to do ever so often"""
raw_data_futures = fetch_data_for_key.map(keys)
process_data.map(raw_data_futures)
if __name__ == "__main__":
work_i_need_to_do_sometimes.serve(
name="sporradic-deployment",
parameters={"keys": ["foo", "bar", "baz"]}
)
The parameters
attribute demonstrates how you can provide a dict with the key "keys"
that takes a list of strings as its value. This deployment will then be capable of being triggered with different sets of keys according to this schema.
View the source code (serve.py)Nate
03/06/2024, 5:55 AMIn [1]: from prefect import flow
In [2]: @flow
...: def foo(x: int, y: str, z: bool = True) -> None:
...: pass
...:
In [3]: foo.parameters.schema()
Out[3]:
{'title': 'ParameterSchema',
'description': 'Simple data model corresponding to an OpenAPI `Schema`.',
'type': 'object',
'properties': {'title': {'title': 'Title',
'default': 'Parameters',
'enum': ['Parameters'],
'type': 'string'},
'type': {'title': 'Type',
'default': 'object',
'enum': ['object'],
'type': 'string'},
'properties': {'title': 'Properties', 'type': 'object'},
'required': {'title': 'Required',
'type': 'array',
'items': {'type': 'string'}},
'definitions': {'title': 'Definitions', 'type': 'object'}}}