alex
01/06/2023, 5:14 PMalex
01/06/2023, 5:15 PMfrom prefect import flow, task
from dataclasses import dataclass
from prefect.deployments import Deployment
@dataclass
class FlowBuildSpecs:
name: str
do_double: bool
do_square: bool
@task
def double(x):
return 2 * x
@task
def square(x):
return x * x
def get_flow(specs):
@flow(name=specs.name)
def my_flow(x=5):
val = x
if specs.do_double:
val = double(val)
if specs.do_square:
val = square(val)
return val
return my_flow
configs = [
FlowBuildSpecs(name="a", do_double=True, do_square=True),
FlowBuildSpecs(name="b", do_double=True, do_square=False),
]
for configs in configs[:1]:
flow = get_flow(configs)
deployment = Deployment.build_from_flow(
flow=flow,
name="local",
tags=["test"],
work_queue_name="test",
)
_id = deployment.apply()
print(f"Deployment ID: {_id}")
alex
01/06/2023, 5:15 PMprefect.exceptions.MissingFlowError: Flow function with name 'my_flow' not found in 'minimal_flow.py'.
alex
01/06/2023, 5:16 PMKalise Richmond
01/06/2023, 7:46 PMalex
01/06/2023, 8:00 PMSamuel Hinton
03/01/2023, 11:05 PMalex
03/01/2023, 11:21 PMSamuel Hinton
03/01/2023, 11:48 PM