Cosmos Configuration Reference (Fusion)

Cosmos Configuration Reference (Fusion)

This reference covers Cosmos configuration for dbt Fusion projects. Fusion only supports ExecutionMode.LOCAL with Snowflake or Databricks warehouses.

Table of Contents

• ProfileConfig: Warehouse Connection
• operator_args Configuration
• Airflow 3 Compatibility

ProfileConfig: Warehouse Connection

Supported ProfileMapping Classes (Fusion)

Note: Fusion currently only supports Snowflake and Databricks (public beta).

Option A: Airflow Connection + ProfileMapping (Recommended)

from cosmos import ProfileConfig
from cosmos.profiles import SnowflakeUserPasswordProfileMapping

_profile_config = ProfileConfig(
    profile_name="default",  # REQUIRED
    target_name="dev",  # REQUIRED
    profile_mapping=SnowflakeUserPasswordProfileMapping(
        conn_id="snowflake_default",  # REQUIRED
        profile_args={"schema": "my_schema"},  # OPTIONAL
    ),
)

Databricks example:

from cosmos import ProfileConfig
from cosmos.profiles import DatabricksTokenProfileMapping

_profile_config = ProfileConfig(
    profile_name="default",
    target_name="dev",
    profile_mapping=DatabricksTokenProfileMapping(
        conn_id="databricks_default",
    ),
)

Option B: Existing profiles.yml File

CRITICAL: Do not hardcode secrets in profiles.yml; use environment variables.

from cosmos import ProfileConfig

_profile_config = ProfileConfig(
    profile_name="my_profile",  # REQUIRED: must match profiles.yml
    target_name="dev",  # REQUIRED: must match profiles.yml
    profiles_yml_filepath="/path/to/profiles.yml",  # REQUIRED
)

operator_args Configuration

The operator_args dict accepts parameters passed to Cosmos operators:

Example Configuration

_operator_args = {
    # BaseOperator params
    "retries": 3,

    # Cosmos-specific params
    "install_deps": False,  # if deps precomputed
    "full_refresh": False,  # for incremental models
    "quiet": True,  # only log errors
}

Passing dbt vars at Runtime (XCom / Params)

Use operator_args["vars"] to pass values from upstream tasks or Airflow params:

# Pull from upstream task via XCom
_operator_args = {
    "vars": '{"my_department": "{{ ti.xcom_pull(task_ids='pre_dbt', key='return_value') }}"}',
}

# Pull from Airflow params (for manual runs)
@dag(params={"my_department": "Engineering"})
def my_dag():
    dbt = DbtTaskGroup(
        # ...
        operator_args={
            "vars": '{"my_department": "{{ params.my_department }}"}',
        },
    )

Airflow 3 Compatibility

Import Differences

Asset/Dataset URI Format Change

Cosmos ≤1.9 (Airflow 2 Datasets):

postgres://0.0.0.0:5434/postgres.public.orders

Cosmos ≥1.10 (Airflow 3 Assets):

postgres://0.0.0.0:5434/postgres/public/orders

CRITICAL: If you have downstream DAGs scheduled on Cosmos-generated datasets and are upgrading to Airflow 3, update the asset URIs to the new format.