Greetings all! I am trying to use Prefect Task Secrets (stored in the prefect server) with the DBTShellTask, but running into some issues. I want to use secrets to set the

db_kwargs

account, etc

dbt = DbtShellTask(
    return_all=True,
    profile_name="test",
    environment="dev",
    overwrite_profiles=True,
    log_stdout=True,
    helper_script="cd dbtProject",
    log_stderr=True,
    dbt_kwargs={
        "type": "snowflake",
        "account": "${ACCOUNT}",
         ...

I then retrieve the account name as a prefect secret, and set it as an env variable called ACCOUNT when I run the flow:

with Flow() as flow:
        account = PrefectSecret("snowflake-account")
        parse = dbt(
            command=f"dbt run --models ./models/{database}",
            env={"ACCOUNT": account}
        )

I have used this approach with ShellTasks, and it works just fine. However, it is not working with DBTShellTasks. Instead of using the value of the Prefect Secret for the account name, it is trying to connect me to snowflake with "${ACCOUNT}" as the account name. Any ideas?? Thanks in advance for the help 😁

Hey @Chris Leber, the DBTShellTask call here is resolved during build time whereas the flow block is resolved during run time. What you need to do is remove the dbt kwargs from the top and instead pass it like

with Flow() as flow:
        account = PrefectSecret("snowflake-account")
        parse = dbt(
            command=f"dbt run --models ./models/{database}",
            dbt_kwargs={
                "type": "snowflake",
                "account": account,
                 ...

This will pass the value of the secret at runtime

That did the trick! Thanks @Kevin Kho so much for the help. Makes sense too - need to get my head fully wrapped around build and run being decoupled steps.