I'm new to Prefect and I'm having difficulty using/understanding arguments with the

DbtShellTask

. Here's some mock code:

from prefect.tasks.dbt.dbt import DbtShellTask

dbt_shell_task = DbtShellTask() 

with Flow("nightly_dbt_flow") as flow:
    env = Parameter('env', default='dev')
    run_task = dbt_shell_task(command='dbt run',
                              environment='dev')

flow.run()

I receive a TypeError that the keyword argument

environment

doesn't exist:

Traceback (most recent call last):
  File "nightly_dbt_flow.py", line 50, in <module>
    environment='dev')
  File "/Users/daniel/git-repos/bi-prefect/.python/lib/python3.7/site-packages/prefect/core/task.py", line 662, in __call__
    *args, mapped=mapped, upstream_tasks=upstream_tasks, flow=flow, **kwargs
  File "/Users/daniel/git-repos/bi-prefect/.python/lib/python3.7/site-packages/prefect/core/task.py", line 702, in bind
    callargs = dict(signature.bind(*args, **kwargs).arguments)  # type: Dict
  File "/Users/daniel/.pyenv/versions/3.7.10/lib/python3.7/inspect.py", line 3015, in bind
    return args[0]._bind(args[1:], kwargs)
  File "/Users/daniel/.pyenv/versions/3.7.10/lib/python3.7/inspect.py", line 3006, in _bind
    arg=next(iter(kwargs))))
TypeError: got an unexpected keyword argument 'environment'

Why would

environment

cause an error but not

command

? When I remove

environment

the task does get attempted without a

TypeError

.

environment

and

command

are both arguments defined in the

DbtShellTask

class. I'm on prefect version

1.2.0

.

Hey @Daniel!

environment

can only be defined when instantiating an instance of

DbtShellTask

. You would pass it in like this:

dbt_shell_task = DbtShellTask(environment="dev")

What is it that makes

environment

only able to be defined on instantiation?

I think that’s just how the current task is written, but it can definitely be fixed easily if you’re interested in making a PR.

why is it important to you for it to be defined in the run method? do you pass the environment name from some upstream task?

Thank you for the link to the code. That is helpful.

That's what I thought. You can either contribute the addition to the run method or pass environment via task_args or change the DBT host to a production host using the dbt_kwargs