https://prefect.io logo
Join Slack
Powered by
Does anyone have experience passing in a task as a...
# ask-community
j
Does anyone have experience passing in a task as a parameter to a deployment? I have a deployment being created using a flow. I'm passing in a Task object within a list as a parameter during the Deployment build process. But I'm running into this error:
Copy code
deployment = Deployment.build_from_flow(
...
parameters={"foo_tasks": [FooTask]}
Copy code
Object of type 'Task' is not JSON serializable
n
this is not a pattern we typically encourage, can you explain your motivation to do this?
j
there is probably a better way to do this but it allows me to dynamically run different kinds of tasks and load that as parameters to the deployment So a scenario might be if there is a postmates api and a grub hub api I can dynamically run the postmates task or the grub hub task by loading the task as a parameter
Copy code
parameters = {"orders": [{"order_number": "2", "api_request": GrubhubTask}, {"order_number": "3", "api_request": "PostmatesTask}] }
I might be wrong in this assumption, otherwise I would need to make a deployment just for grubhub order and another one for postmates order in this example
unless there is a way to pass the task name in as a parameters and somehow that Task gets picked up downstream
n
how are you calling the appropriate task within the flow run context?
it feels like you should be able to have some logic that uses the flow run param values to call the right task
j
In my flow I am able to loop over the "orders" parameter and do this:
Copy code
futures = []
for order in orders:
    futures.append(order.api_request.submit())
to oversimplify, that returns a list of futures
the task runner async performs these tasks
I guess I could have a request_type: str and have a bunch of if statements to do this:
Copy code
futures = []
for order in orders:
    if order_type == 'grubhub'
        futures.append(grubhub_task.submit(**order.params))
    elif order_type == 'postmates'
        futures.append(postmates_task.submit(**order.params))
that way the deployment parameters can be serialized since the order_type is just a string
n
i see, perhaps instead of passing the literal task object as a param you could
Copy code
In [10]: from prefect import flow, task
    ...:
    ...: @task
    ...: def call_grubhub_api():
    ...:     print("calling grubhub")
    ...:
    ...: @task
    ...: def call_postmates_api():
    ...:     print("calling postmates")
    ...:
    ...:
    ...: api_request_tasks = {
    ...:    "grubhub": call_grubhub_api,
    ...:    "postmates": call_postmates_api
    ...: }
and then
Copy code
parameters = {"orders": [{"order_number": "2", "api_request": "grubhub"}, {"order_number": "3", "api_request": "postmates"}] }
which is basically your if statement solution above, just without needing a bunch of if statements
j
yup that makes a lot of sense!
thanks for letting me spring board off you
n
sure thing!
2 Views
Open in Slack
PreviousNext
Powered by