Skip to content

prefect_dbt.cli.configs.snowflake

Module containing models for Snowflake configs

SnowflakeTargetConfigs

Bases: BaseTargetConfigs

Target configs contain credentials and settings, specific to Snowflake. To find valid keys, head to the Snowflake Profile page.

Attributes:

Name Type Description
connector SnowflakeConnector

The connector to use.

Examples:

Load stored SnowflakeTargetConfigs: 

from prefect_dbt.cli.configs import SnowflakeTargetConfigs

snowflake_target_configs = SnowflakeTargetConfigs.load("BLOCK_NAME")

Instantiate SnowflakeTargetConfigs. 

from prefect_dbt.cli.configs import SnowflakeTargetConfigs
from prefect_snowflake.credentials import SnowflakeCredentials
from prefect_snowflake.database import SnowflakeConnector

credentials = SnowflakeCredentials(
    user="user",
    password="password",
    account="account.region.aws",
    role="role",
)
connector = SnowflakeConnector(
    schema="public",
    database="database",
    warehouse="warehouse",
    credentials=credentials,
)
target_configs = SnowflakeTargetConfigs(
    connector=connector,
    extras={"retry_on_database_errors": True},
)

Source code in prefect_dbt/cli/configs/snowflake.py 
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
class SnowflakeTargetConfigs(BaseTargetConfigs):
    """
    Target configs contain credentials and
    settings, specific to Snowflake.
    To find valid keys, head to the [Snowflake Profile](
    https://docs.getdbt.com/reference/warehouse-profiles/snowflake-profile)
    page.

    Attributes:
        connector: The connector to use.

    Examples:
        Load stored SnowflakeTargetConfigs:
        ```python
        from prefect_dbt.cli.configs import SnowflakeTargetConfigs

        snowflake_target_configs = SnowflakeTargetConfigs.load("BLOCK_NAME")
        ```

        Instantiate SnowflakeTargetConfigs.
        ```python
        from prefect_dbt.cli.configs import SnowflakeTargetConfigs
        from prefect_snowflake.credentials import SnowflakeCredentials
        from prefect_snowflake.database import SnowflakeConnector

        credentials = SnowflakeCredentials(
            user="user",
            password="password",
            account="account.region.aws",
            role="role",
        )
        connector = SnowflakeConnector(
            schema="public",
            database="database",
            warehouse="warehouse",
            credentials=credentials,
        )
        target_configs = SnowflakeTargetConfigs(
            connector=connector,
            extras={"retry_on_database_errors": True},
        )
        ```
    """

    _block_type_name = "dbt CLI Snowflake Target Configs"
    _logo_url = "https://images.ctfassets.net/gm98wzqotmnx/5zE9lxfzBHjw3tnEup4wWL/9a001902ed43a84c6c96d23b24622e19/dbt-bit_tm.png?h=250"  # noqa
    _documentation_url = "https://prefecthq.github.io/prefect-dbt/cli/configs/snowflake/#prefect_dbt.cli.configs.snowflake.SnowflakeTargetConfigs"  # noqa

    type: Literal["snowflake"] = Field(
        default="snowflake", description="The type of the target configs."
    )
    schema_: Optional[str] = Field(
        default=None,
        alias="schema",
        description="The schema to use for the target configs.",
    )
    connector: SnowflakeConnector = Field(
        default=..., description="The connector to use."
    )

    def get_configs(self) -> Dict[str, Any]:
        """
        Returns the dbt configs specific to Snowflake profile.

        Returns:
            A configs JSON.
        """
        all_configs_json = super().get_configs()

        # decouple prefect-snowflake from prefect-dbt
        # by mapping all the keys dbt snowflake accepts
        # https://docs.getdbt.com/reference/warehouse-setups/snowflake-setup
        rename_keys = {
            # dbt
            "type": "type",
            "schema": "schema",
            "threads": "threads",
            # general
            "account": "account",
            "user": "user",
            "role": "role",
            "database": "database",
            "warehouse": "warehouse",
            # user and password
            "password": "password",
            # duo mfa / sso
            "authenticator": "authenticator",
            # key pair
            "private_key_path": "private_key_path",
            "private_key_passphrase": "private_key_passphrase",
            # optional
            "client_session_keep_alive": "client_session_keep_alive",
            "query_tag": "query_tag",
            "connect_retries": "connect_retries",
            "connect_timeout": "connect_timeout",
            "retry_on_database_errors": "retry_on_database_errors",
            "retry_all": "retry_all",
        }
        configs_json = {}
        extras = self.extras or {}
        for key in all_configs_json.keys():
            if key not in rename_keys and key not in extras:
                # skip invalid keys, like fetch_size + poll_frequency_s
                continue
            # rename key to something dbt profile expects
            dbt_key = rename_keys.get(key) or key
            configs_json[dbt_key] = all_configs_json[key]
        return configs_json

get_configs

Returns the dbt configs specific to Snowflake profile.

Returns:

Type Description
Dict[str, Any]

A configs JSON.
Source code in prefect_dbt/cli/configs/snowflake.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def get_configs(self) -> Dict[str, Any]:
    """
    Returns the dbt configs specific to Snowflake profile.

    Returns:
        A configs JSON.
    """
    all_configs_json = super().get_configs()

    # decouple prefect-snowflake from prefect-dbt
    # by mapping all the keys dbt snowflake accepts
    # https://docs.getdbt.com/reference/warehouse-setups/snowflake-setup
    rename_keys = {
        # dbt
        "type": "type",
        "schema": "schema",
        "threads": "threads",
        # general
        "account": "account",
        "user": "user",
        "role": "role",
        "database": "database",
        "warehouse": "warehouse",
        # user and password
        "password": "password",
        # duo mfa / sso
        "authenticator": "authenticator",
        # key pair
        "private_key_path": "private_key_path",
        "private_key_passphrase": "private_key_passphrase",
        # optional
        "client_session_keep_alive": "client_session_keep_alive",
        "query_tag": "query_tag",
        "connect_retries": "connect_retries",
        "connect_timeout": "connect_timeout",
        "retry_on_database_errors": "retry_on_database_errors",
        "retry_all": "retry_all",
    }
    configs_json = {}
    extras = self.extras or {}
    for key in all_configs_json.keys():
        if key not in rename_keys and key not in extras:
            # skip invalid keys, like fetch_size + poll_frequency_s
            continue
        # rename key to something dbt profile expects
        dbt_key = rename_keys.get(key) or key
        configs_json[dbt_key] = all_configs_json[key]
    return configs_json
Was this page helpful?