Ken Nguyen
04/17/2022, 7:21 PMwith 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")
run_dbt.map(
command=str(unmapped(dbt_command)) + flags_to_add,
)
But got the following error:
TypeError: can only concatenate str (not "list") to str
Kevin Kho
Anna Geller
from prefect import task, Parameter, Flow
from typing import List
from prefect.tasks.dbt.dbt import DbtShellTask
run_dbt = DbtShellTask(...)
@task
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)
run_dbt.map(command=dbt_commands)
I think you don't need unmapped
because you want to iterate over this command in your mapping, correct?Ken Nguyen
04/17/2022, 8:31 PM