Hey All! I am trying to run a

DbtShellTask

with a dynamic set of

dbt_kwargs

that are defined at run time. The task works when I define the

dbt_kwargs

in the task initialization, but not when they are provided at runtime.

from prefect import task, Flow, Parameter
from prefect.tasks.dbt import DbtShellTask

@task
def generate_dbt_kwargs_dict(project_id):
    dbt_kwargs_dict = {
        'type': 'bigquery',
        'method': 'service-account',
        'keyfile': 'key.json',
        'project': project_id,
        'dataset': 'derived_data',
        'location': 'EU',
    }
    return dbt_kwargs_dict

dbt_run_task = DbtShellTask(
        name='dbt_run_task',
        command='dbt run',
        profile_name='default',
        environment='test',
        overwrite_profiles=True,
        profiles_dir=profiles_path,
    )

with Flow(name="dbt_flow") as f:
    project_id = Parameter("project_id")
    dbt_kwargs_dict = generate_dbt_kwargs_dict(project_id)
    run_task = dbt_run_task(dbt_kwargs=dbt_kwargs_dict)

out = f.run(project_id='bq-project')

This returns a prefect error:

ERROR - prefect.TaskRunner | Unexpected error: TypeError("'NoneType' object is not a mapping",)

I assume this means from here that the

dbt_kwargs

dictionary is

None

but I can’t see why, since I am passing the dictionary from the previous task… Does anyone have an idea about this? Thanks!

Thanks for letting us know you figured it out @Nate Joselson - hopefully your question might help others who encounter a similar issue.

I kind of think that the documentation or the task default arguments should be changed. Either change the documentation to say that it requires an empty dict at initialization but that this dict can be modified at runtime, or change the default arguments to not be

None

so that the dictionary modification I mentioned above works as expected. Thanks so much!

I opened an issue here @Nate Joselson - feel free to check it out and comment.

Thanks so much!