Skip to content

prefect_dbt.cli.credentials

Module containing credentials for interacting with dbt CLI

DbtCliProfile

Bases: Block

Profile for use across dbt CLI tasks and flows.

Attributes:

Name Type Description
name str

Profile name used for populating profiles.yml.
target str

The default target your dbt project will use.
target_configs TargetConfigs

Target configs contain credentials and settings, specific to the warehouse you're connecting to. To find valid keys, head to the Available adapters page and click the desired adapter's "Profile Setup" hyperlink.
global_configs GlobalConfigs

Global configs control things like the visual output of logs, the manner in which dbt parses your project, and what to do when dbt finds a version mismatch or a failing model. Valid keys can be found here.

Examples:

Load stored dbt CLI profile: 

from prefect_dbt.cli import DbtCliProfile
dbt_cli_profile = DbtCliProfile.load("BLOCK_NAME").get_profile()

Get a dbt Snowflake profile from DbtCliProfile by using SnowflakeTargetConfigs: 

from prefect_dbt.cli import DbtCliProfile
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
)
dbt_cli_profile = DbtCliProfile(
    name="jaffle_shop",
    target="dev",
    target_configs=target_configs,
)
profile = dbt_cli_profile.get_profile()

Get a dbt Redshift profile from DbtCliProfile by using generic TargetConfigs: 

from prefect_dbt.cli import DbtCliProfile
from prefect_dbt.cli.configs import GlobalConfigs, TargetConfigs

target_configs_extras = dict(
    host="hostname.region.redshift.amazonaws.com",
    user="username",
    password="password1",
    port=5439,
    dbname="analytics",
)
target_configs = TargetConfigs(
    type="redshift",
    schema="schema",
    threads=4,
    extras=target_configs_extras
)
dbt_cli_profile = DbtCliProfile(
    name="jaffle_shop",
    target="dev",
    target_configs=target_configs,
)
profile = dbt_cli_profile.get_profile()

Source code in prefect_dbt/cli/credentials.py 
 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
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
class DbtCliProfile(Block):
    """
    Profile for use across dbt CLI tasks and flows.

    Attributes:
        name (str): Profile name used for populating profiles.yml.
        target (str): The default target your dbt project will use.
        target_configs (TargetConfigs): Target configs contain credentials and
            settings, specific to the warehouse you're connecting to.
            To find valid keys, head to the [Available adapters](
            https://docs.getdbt.com/docs/available-adapters) page and
            click the desired adapter's "Profile Setup" hyperlink.
        global_configs (GlobalConfigs): Global configs control
            things like the visual output of logs, the manner
            in which dbt parses your project, and what to do when
            dbt finds a version mismatch or a failing model.
            Valid keys can be found [here](
            https://docs.getdbt.com/reference/global-configs).

    Examples:
        Load stored dbt CLI profile:
        ```python
        from prefect_dbt.cli import DbtCliProfile
        dbt_cli_profile = DbtCliProfile.load("BLOCK_NAME").get_profile()
        ```

        Get a dbt Snowflake profile from DbtCliProfile by using SnowflakeTargetConfigs:
        ```python
        from prefect_dbt.cli import DbtCliProfile
        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
        )
        dbt_cli_profile = DbtCliProfile(
            name="jaffle_shop",
            target="dev",
            target_configs=target_configs,
        )
        profile = dbt_cli_profile.get_profile()
        ```

        Get a dbt Redshift profile from DbtCliProfile by using generic TargetConfigs:
        ```python
        from prefect_dbt.cli import DbtCliProfile
        from prefect_dbt.cli.configs import GlobalConfigs, TargetConfigs

        target_configs_extras = dict(
            host="hostname.region.redshift.amazonaws.com",
            user="username",
            password="password1",
            port=5439,
            dbname="analytics",
        )
        target_configs = TargetConfigs(
            type="redshift",
            schema="schema",
            threads=4,
            extras=target_configs_extras
        )
        dbt_cli_profile = DbtCliProfile(
            name="jaffle_shop",
            target="dev",
            target_configs=target_configs,
        )
        profile = dbt_cli_profile.get_profile()
        ```
    """

    _block_type_name = "dbt CLI Profile"
    _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/credentials/#prefect_dbt.cli.credentials.DbtCliProfile"  # noqa

    name: str = Field(
        default=..., description="Profile name used for populating profiles.yml."
    )
    target: str = Field(
        default=..., description="The default target your dbt project will use."
    )
    target_configs: Union[
        SnowflakeTargetConfigs,
        BigQueryTargetConfigs,
        PostgresTargetConfigs,
        TargetConfigs,
    ] = Field(
        default=...,
        description=(
            "Target configs contain credentials and settings, specific to the "
            "warehouse you're connecting to."
        ),
    )
    global_configs: Optional[GlobalConfigs] = Field(
        default=None,
        description=(
            "Global configs control things like the visual output of logs, the manner "
            "in which dbt parses your project, and what to do when dbt finds a version "
            "mismatch or a failing model."
        ),
    )

    def get_profile(self) -> Dict[str, Any]:
        """
        Returns the dbt profile, likely used for writing to profiles.yml.

        Returns:
            A JSON compatible dictionary with the expected format of profiles.yml.
        """
        profile = {
            "config": self.global_configs.get_configs() if self.global_configs else {},
            self.name: {
                "target": self.target,
                "outputs": {self.target: self.target_configs.get_configs()},
            },
        }
        return profile

get_profile

Returns the dbt profile, likely used for writing to profiles.yml.

Returns:

Type Description
Dict[str, Any]

A JSON compatible dictionary with the expected format of profiles.yml.
Source code in prefect_dbt/cli/credentials.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
def get_profile(self) -> Dict[str, Any]:
    """
    Returns the dbt profile, likely used for writing to profiles.yml.

    Returns:
        A JSON compatible dictionary with the expected format of profiles.yml.
    """
    profile = {
        "config": self.global_configs.get_configs() if self.global_configs else {},
        self.name: {
            "target": self.target,
            "outputs": {self.target: self.target_configs.get_configs()},
        },
    }
    return profile