Is there any way to pass a dictionary to the unmapped() annotation? I am trying to map a create_flow...

Daniel Lomartra

06/03/2022, 1:03 AM

Is there any way to pass a dictionary to the unmapped() annotation? I am trying to map a create_flow_run task and use the task_args parameter to name the flow run.

Kevin Kho

06/03/2022, 1:18 AM

I don’t see why not. What happens when you try it? Could you show me the syntax you used?

Daniel Lomartra

06/03/2022, 1:52 AM

Copy code

update_sync_time = create_flow_run.map(
        flow_name = unmapped("dbt"),
        project_name = unmapped(PROJECT_NAME),
        task_args= unmapped({"name":"Update Sync Time in Database"}),
        parameters=portco
    )
    update_sync_time_wait = wait_for_flow_run.map(
        update_sync_time, raise_final_state=unmapped(True), task_args=unmapped({"name":"Update Sync Time in Database - Wait"})
    )

Daniel Lomartra

06/03/2022, 1:53 AM

Copy code

Traceback (most recent call last):
  File "C:\Users\DLomartra\source\repos\MEP2\prefect-cloud\01_main\flows\parent_flows\dbt-parent.py", line 65, in <module>
    update_sync_time = create_flow_run.map(
  File "C:\Users\DLomartra\source\repos\MEP2\prefect-cloud\01_main\lib\site-packages\prefect\core\task.py", line 810, in map
    task_args = task_args.copy() if task_args else {}
AttributeError: 'unmapped' object has no attribute 'copy'

Kevin Kho

06/03/2022, 1:56 AM

Ah task args is evaluated during build time, but runtime. Unmapped is a task though I think so you get this weird behavior. Based on this, I think I’d do:

Copy code

from prefect.tasks.prefect import create_flow_run

create_flow_run.name = "Update Sync Time in Database"

with Flow(..) as flow:
    create_flow_run.map(...)

Daniel Lomartra

06/03/2022, 4:50 PM

Thanks. In this specific case I have multiple create_flow_run's in the flow, each needing a different name. How would I implement in that case?

Kevin Kho

06/03/2022, 5:02 PM

Maybe you want to template the

task_run_name

instead like this ? You could also copy the task

Copy code

from prefect.tasks.prefect import create_flow_run
from copy import copy

c1 = copy(create_flow_run)
c1.name = ...

🙌 1

Open in Slack

Previous Next

Prefect Community

Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.