Ken Nguyen

04/17/2022, 7:21 PM
If I have 2 parameters, one as a list to be mapped, and one as a value that should be unmapped, how can I concat the mapped parameters onto the unmapped one?
For example I tried something like this:
with Flow("data-quality-tracking-parallel-dbt-run-flow", run_config=RUN_CONFIG, storage=STORAGE, executor=LocalDaskExecutor()) as flow:
    flags_to_add = Parameter('flags_to_add', default=["--flag1", "--flag2"], required=False)
    dbt_command = Parameter('dbt_command', default="dbt run")
        command=str(unmapped(dbt_command)) + flags_to_add,
But got the following error:
TypeError: can only concatenate str (not "list") to str

Kevin Kho

04/17/2022, 7:38 PM
You have to do it in a task, this is a build time vs run time thing. str on unmapped is just a str on a task class so it’s not what you want
:upvote: 1

Anna Geller

04/17/2022, 8:28 PM
I think you need something like this - as Kevin mentioned, a separate task that will generated a list of commands for mapping based on your input parameter values passed as data dependencies:
from prefect import task, Parameter, Flow
from typing import List
from prefect.tasks.dbt.dbt import DbtShellTask

run_dbt = DbtShellTask(...)

def get_full_dbt_commands(
    flags_to_add: List[str], dbt_command: str = "dbt run"
) -> List[str]:
    return [f"{dbt_command} {flag}" for flag in flags_to_add]

with Flow("data-quality-tracking-parallel-dbt-run-flow") as flow:
    dbt_flags = Parameter("flags_to_add", default=["--flag1", "--flag2"])
    dbt_command = Parameter("dbt_command", default="dbt run")
    dbt_commands = get_full_dbt_commands(dbt_flags, dbt_command)
I think you don't need
because you want to iterate over this command in your mapping, correct?

Ken Nguyen

04/17/2022, 8:31 PM
Thank you both!
👍 1