dbt deps

dbt debug

@flow
def trigger_dbt_cli_command_flow():
    with open(SNOWFLAKE_SVC_USER_PRIVATE_KEY_FILE) as key_file:
        connector = SnowflakeConnector(
            database=SNOWFLAKE_DATABASE,
            schema=SNOWFLAKE_SCHEMA,
            warehouse=SNOWFLAKE_WAREHOUSE,
            credentials=SnowflakeCredentials(
                user=SNOWFLAKE_USERNAME,  # users have default roles, so we don't provide role here
                private_key=key_file.read().encode(),
                account=SNOWFLAKE_ACCOUNT,
            ),
        )
    target_configs = SnowflakeTargetConfigs(connector=connector)
    dbt_cli_profile = DbtCliProfile(
        name=APP,
        target=DBT_ENVIRONMENT,
        target_configs=target_configs,
    )
    trigger_dbt_cli_command(
        "dbt deps",
        overwrite_profiles=True,
        dbt_cli_profile=dbt_cli_profile,
        project_dir=DBT_PROJECT_DIRECTORY,
    )
    result = trigger_dbt_cli_command(
        "dbt debug",
        overwrite_profiles=True,
        dbt_cli_profile=dbt_cli_profile,
        project_dir=DBT_PROJECT_DIRECTORY,
    )
    return result

dbt_project.yml

│ 19:50:02  Encountered an error:                                                                                                                                                                                                                   
│ Compilation Error                                                                                                                                                                                                                                 
│   Could not render {{ true if target.name == 'prod' else false }}: 'target' is undefined                                                                                                                                                          │                                                                                                                                                                                                                                
│ 19:50:03.145 | ERROR   | Task run 'trigger_dbt_cli_command-321ca940-0' - Encountered exception during execution:                                                                                                                                  
│ Traceback (most recent call last):                                                                                                                                                                                                                
│   File "/usr/local/lib/python3.9/site-packages/prefect/engine.py", line 1442, in orchestrate_task_run                                                                                                                                             
│     result = await task.fn(*args, **kwargs)                                                                                                                                                                                                       
│   File "/usr/local/lib/python3.9/site-packages/prefect_dbt/cli/commands.py", line 158, in trigger_dbt_cli_command                                                                                                                                 
│     result = await shell_run_command.fn(command=command, **shell_run_command_kwargs)                                                                                                                                                              
│   File "/usr/local/lib/python3.9/site-packages/prefect_shell/commands.py", line 103, in shell_run_command                                                                                                                                         
│     raise RuntimeError(msg)                                                                                                                                                                                                                       
│ RuntimeError: Command failed with exit code 2:                                                                                                                                                                                                    
│   Could not render {{ true if target.name == 'prod' else false }}: 'target' is undefined

dbt_project.yml

RuntimeError: Command failed with exit code 2:
  in _deep_map_render, expected one of (<class 'list'>, <class 'dict'>, <class 'int'>, <class 'float'>, <class 'str'>, <class 'NoneType'>, <class 'bool'>), got <class 'bytes'>

~/.dbt/profiles.yml

app:
  outputs:
    qa:
      account: XXXXXX.us-east-1
      authenticator: snowflake
      database: qa
      private_key: !!binary |
        XXX (redacted) XXX
      schema: public
      threads: 4
      type: snowflake
      user: svc_user_qa
      warehouse: qa_qh
  target: qa

private_key

private_key_path: /path/to/key.p8
      private_key_passphrase: **redacted**

SnowflakeCredentials

SnowflakeConnector

DbtCliProfile

SnowflakeCredentials

class SnowflakePrivateKeyCredentials(SnowflakeCredentials):
    """
    SnowflakeCredentials expects private key bytes, which doesn't play
    nicely with dbt. To get around this, we use a custom model that has the
    right fields.
    """

    _block_type_name = "Snowflake Private Key Credentials"
    _block_type_slug = "snowflake-private-key-credentials"
    private_key_path: Optional[str] = Field(
        default=None, description="The path to the private key"
    )
    private_key_passphrase: Optional[SecretStr] = Field(
        default=None, description="The passphrase to decrypt the private key"
    )

snowflake.connector.errors.ProgrammingError: 251006: Password is empty

prefect-snowflake==0.2.3

configs_json = super().get_configs()
        rename_keys = {
            "account",
            "user",
            "password",
            "private_key": "private_key_path",
            "authenticator",
            "database",
            "warehouse",
            "schema",
        }
        for key in configs_json.keys():
            if key not in rename_keys:
                configs_json.pop(key)

class SnowflakePrivateKeyCredentials(SnowflakeCredentials):
    """
    SnowflakeCredentials expects private key bytes, which doesn't play
    nicely with dbt. To get around this, we use a custom model that has the
    right fields.
    """

    _block_type_name = "Snowflake Private Key Credentials"
    _block_type_slug = "snowflake-private-key-credentials"
    private_key_path: Optional[str] = Field(
        default=None, description="The path to the private key"
    )
    private_key_passphrase: Optional[SecretStr] = Field(
        default=None, description="The passphrase to decrypt the private key"
    )

def run_dbt_command(command: str):
    """
    Run the provided dbt command using the correct snowflake/dbt configuration.
    This function is a wrapper for the `trigger_dbt_cli_command` prefect-dbt task.
    """
    connector = SnowflakeConnector(
        database=SNOWFLAKE_DATABASE,
        schema=SNOWFLAKE_SCHEMA,
        warehouse=SNOWFLAKE_WAREHOUSE,
        credentials=SnowflakePrivateKeyCredentials(
            user=SNOWFLAKE_USERNAME,
            authenticator="snowflake",
            private_key_path=SNOWFLAKE_SVC_USER_PRIVATE_KEY_FILE,
            private_key_passphrase=get_ssm_parameter(
                f"/snowflake/svc_user/{SSM_ENV}/PRIVATE_KEY_PASSPHRASE"
            ),
            account=SNOWFLAKE_ACCOUNT,
        ),
    )
    target_configs = SnowflakeTargetConfigs(connector=connector)
    dbt_cli_profile = DbtCliProfile(
        name=APP,
        target=DBT_ENVIRONMENT,
        target_configs=target_configs,
    )
    return trigger_dbt_cli_command(
        command=command,
        overwrite_profiles=True,
        dbt_cli_profile=dbt_cli_profile,
        project_dir=DBT_PROJECT_DIRECTORY,
    )

DbtCliProfile

password

private_key_passphrase

Do not provide both password and private_key_passphrase; specify private_key_passphrase only instead.

profiles.yml

config: {}
myproj:
  outputs:
    dev:
      schema: dv
      threads: 8
      type: snowflake
  target: dev

from prefect import flow

from prefect_dbt.cli.credentials import DbtCliProfile
from prefect_dbt.cli.commands import trigger_dbt_cli_command


@flow
def trigger_dbt_cli_commands_flow():
    dbt_cli_profile = DbtCliProfile.load("dbtprofileblock")

    trigger_kwargs = dict(
        profiles_dir=".",
        overwrite_profiles=True,
        dbt_cli_profile=dbt_cli_profile,
    )

    trigger_dbt_cli_command("dbt deps", **trigger_kwargs)

    result = trigger_dbt_cli_command("dbt debug", **trigger_kwargs)

    return result


trigger_dbt_cli_commands_flow()

from prefect_snowflake.credentials import SnowflakeCredentials
from prefect_snowflake.database import SnowflakeConnector

from prefect_dbt.cli.credentials import DbtCliProfile
from prefect_dbt.cli.configs import SnowflakeTargetConfigs


if __name__ == "__main__":
    credentials = SnowflakeCredentials(
        user="user",
        password="password",
        account="account.region.aws",
        role="role",
    )
    credentials.save("sf-cred", overwrite=True)

    connector = SnowflakeConnector(
        schema="public",
        database="database",
        warehouse="warehouse",
        credentials=SnowflakeCredentials.load("sf-cred"),
    )
    connector.save("sf-conn", overwrite=True)

    target_configs = SnowflakeTargetConfigs(
        connector=SnowflakeConnector.load("sf-conn"), schema="TEST"
    )
    target_configs.save("dbt-targetconf", overwrite=True)

    dbt_cli_profile = DbtCliProfile(
        name="jaffle_shop",
        target="dev",
        target_configs=SnowflakeTargetConfigs.load("dbt-targetconf"),
    )
    dbt_cli_profile.save("example", overwrite=True)

    dbt_cli_profile = DbtCliProfile.load("example").get_profile()

    print(dbt_cli_profile)

{'config': {}, 'jaffle_shop': {'target': 'dev', 'outputs': {'dev': {'type': 'snowflake', 'schema': 'TEST', 'threads': 4}}}}