Daniel Lomartra
06/05/2022, 12:01 AMwith Flow(
name=FLOW_CONSTANTS["flow_name"],
storage=STORAGE,
run_config = DockerRun(
image=DOCKER_IMAGE,
labels=FLOW_LABELS
)
) as flow:
dbt_job = Parameter(
name='dbt_job',
default=[
{
"name": "Job Step Name 1", "command": r"Arbitrary dbt CLI command #1"
},
{
"name": "Job Step Name 2", "command": r"Arbitrary dbt CLI command #2"
}
]
)
dbt.map(
command = dbt_job['command'],
task_args= {"name": dbt_job['name']}
)
Anna Geller
Daniel Lomartra
06/05/2022, 1:46 PMTraceback (most recent call last):
File "C:\Users\DLomartra\source\repos\MEP2\prefect-cloud\01_main\lib\site-packages\prefect\cli\build_register.py", line 133, in load_flows_from_script
namespace = runpy.run_path(abs_path, run_name="<flow>")
File "C:\Users\MarlinAdmin\AppData\Local\Programs\Python\Python39\lib\runpy.py", line 268, in run_path
return _run_module_code(code, init_globals, run_name,
File "C:\Users\MarlinAdmin\AppData\Local\Programs\Python\Python39\lib\runpy.py", line 97, in _run_module_code
_run_code(code, mod_globals, init_globals,
File "C:\Users\MarlinAdmin\AppData\Local\Programs\Python\Python39\lib\runpy.py", line 87, in _run_code
exec(code, run_globals)
File "C:\Users\DLomartra\source\repos\MEP2\prefect-cloud\01_main\flow\orchestration\test.py", line 108, in <module>
dbt.map(
File "C:\Users\DLomartra\source\repos\MEP2\prefect-cloud\01_main\lib\site-packages\prefect\core\task.py", line 813, in map
return new.bind(
File "C:\Users\DLomartra\source\repos\MEP2\prefect-cloud\01_main\lib\site-packages\prefect\core\task.py", line 754, in bind
self.set_dependencies(
File "C:\Users\DLomartra\source\repos\MEP2\prefect-cloud\01_main\lib\site-packages\prefect\core\task.py", line 854, in set_dependencies
flow.set_dependencies(
File "C:\Users\DLomartra\source\repos\MEP2\prefect-cloud\01_main\lib\site-packages\prefect\core\flow.py", line 985, in set_dependencies
self.add_task(task)
File "C:\Users\DLomartra\source\repos\MEP2\prefect-cloud\01_main\lib\site-packages\prefect\core\flow.py", line 535, in add_task
self.slugs[task] = task.slug or self._generate_task_slug(task)
File "C:\Users\DLomartra\source\repos\MEP2\prefect-cloud\01_main\lib\site-packages\prefect\core\flow.py", line 503, in _generate_task_slug
prefix = "-".join(parts)
TypeError: sequence item 0: expected str instance, GetItem found
Kevin Kho
Anna Geller