Cosmos Configuration Reference (Core)

Cosmos Configuration Reference (Core)

This reference covers detailed Cosmos configuration for dbt Core projects.

Table of Contents

• Execution Modes (ExecutionConfig)
• ProfileConfig: Warehouse Connection
• Testing Behavior (RenderConfig)
• operator_args Configuration
• ProjectConfig Options
• Airflow 3 Compatibility

Execution Modes (ExecutionConfig)

WATCHER Mode (Experimental, Fastest)

Known limitations:

• Only DbtSeedWatcherOperator, DbtSnapshotWatcherOperator, DbtRunWatcherOperator supported
• TestBehavior.AFTER_EACH NOT supported
• Retrying DbtProducerWatcherOperator NOT supported
• Sensors run synchronously (occupy worker slot)

from cosmos import ExecutionConfig, ExecutionMode

_execution_config = ExecutionConfig(
    execution_mode=ExecutionMode.WATCHER,
)

Reference: https://astronomer.github.io/astronomer-cosmos/getting_started/watcher-execution-mode.html

LOCAL Mode (Default)

from cosmos import ExecutionConfig, ExecutionMode, InvocationMode

# Option A: dbt in Airflow env
_execution_config = ExecutionConfig(
    execution_mode=ExecutionMode.LOCAL,
    invocation_mode=InvocationMode.DBT_RUNNER,
)

# Option B: dbt in separate venv baked into image
_execution_config = ExecutionConfig(
    execution_mode=ExecutionMode.LOCAL,
    invocation_mode=InvocationMode.SUBPROCESS,
    dbt_executable_path="/path/to/venv/bin/dbt",
)

VIRTUALENV Mode

from cosmos import ExecutionConfig, ExecutionMode

_execution_config = ExecutionConfig(
    execution_mode=ExecutionMode.VIRTUALENV,
    virtualenv_dir="/path/to/persistent/cache",
)

_operator_args = {
    "py_system_site_packages": False,
    "py_requirements": ["dbt-<adapter>==<version>"],
    "install_deps": True,
}

AIRFLOW_ASYNC Mode (BigQuery Only)

CRITICAL: BigQuery only, Airflow ≥2.8 required.

Required setup:

1. Install: apache-airflow-providers-google
2. Set env vars:
   • AIRFLOW__COSMOS__REMOTE_TARGET_PATH = gs://bucket/target_dir/
   • AIRFLOW__COSMOS__REMOTE_TARGET_PATH_CONN_ID = connection ID

from cosmos import ExecutionConfig, ExecutionMode

_execution_config = ExecutionConfig(
    execution_mode=ExecutionMode.AIRFLOW_ASYNC,
    async_py_requirements=["dbt-bigquery==<version>"],
)

_operator_args = {
    "location": "US",
    "install_deps": True,
}

Reference: https://astronomer.github.io/astronomer-cosmos/getting_started/async-execution-mode.html

Containerized Modes

Available: DOCKER, KUBERNETES, AWS_EKS, AZURE_CONTAINER_INSTANCE, GCP_CLOUD_RUN_JOB, AWS_ECS.

CRITICAL: MUST use dbt_manifest load mode.

from cosmos import ExecutionConfig, ExecutionMode, RenderConfig, LoadMode

_execution_config = ExecutionConfig(
    execution_mode=ExecutionMode.KUBERNETES,
    dbt_project_path="/path/to/dbt/project/in/image",
)

_render_config = RenderConfig(
    load_method=LoadMode.DBT_MANIFEST,
)

_operator_args = {
    "image": "dbt-jaffle-shop:1.0.0",
}

ProfileConfig: Warehouse Connection

ProfileMapping Classes by Warehouse

Full list: https://astronomer.github.io/astronomer-cosmos/profiles/index.html

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
    ),
)

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
)

Per-Node Profile Override

Override profile for individual nodes via dbt_project.yml:

# In dbt_project.yml or models/*.yml
version: 2

models:
  - name: my_model
    meta:
      cosmos:
        profile_config:
          profile_name: other_profile
          target_name: prod
          profile_mapping:
            conn_id: other_connection
            profile_args:
              schema: prod

Testing Behavior (RenderConfig)

TestBehavior Options

NOTE: Cosmos default (AFTER_EACH) differs from dbt CLI default (AFTER_ALL).

Multi-Parent Test Handling

If a test depends on multiple models, AFTER_EACH may fail because not all parent models are materialized yet.

Solution: Set should_detach_multiple_parents_tests=True to run multi-parent tests only after all their parents complete.

from cosmos import RenderConfig, TestBehavior

_render_config = RenderConfig(
    test_behavior=TestBehavior.AFTER_EACH,  # default
    # should_detach_multiple_parents_tests=True,  # for multi-parent tests
)

test_indirect_selection (For Subset Runs)

When running only part of a project (select/exclude), control which tests run. Set in ExecutionConfig:

from cosmos import ExecutionConfig, TestIndirectSelection

_execution_config = ExecutionConfig(
    test_indirect_selection=TestIndirectSelection.CAUTIOUS,
)

on_warning_callback

Execute a function when dbt tests generate warnings (works with local, virtualenv, kubernetes execution modes):

from airflow.utils.context import Context

def warning_callback(context: Context):
    tests = context.get("test_names")
    results = context.get("test_results")
    # Send to Slack, email, etc.

my_dag = DbtDag(
    # ...
   
)

operator_args Configuration

The operator_args dict accepts four categories of parameters:

Parameter Categories

Example Configuration

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

    # Cosmos-specific params
    "install_deps": False,  # if deps precomputed
    "full_refresh": False,  # for incremental models
    "quiet": True,  # only log errors
    "fail_fast": True,  # exit immediately on failure

    # Container params (for containerized execution)
    "image": "my-dbt-image:latest",
    "namespace": "airflow",
}

Passing dbt vars at Runtime (XCom / Params)

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

WARNING: operator_args["vars"] overrides ALL vars in ProjectConfig.dbt_vars.

# 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 }}"}',
        },
    )

Per-Node Operator Overrides

Override task parameters for individual nodes via dbt_project.yml:

# In dbt_project.yml or models/*.yml
version: 2

models:
  - name: my_model
    meta:
      cosmos:
        operator_kwargs:
          retries: 10
          pool: "high_priority_pool"

ProjectConfig Options

Required Parameters

Optional Parameters

WARNING: If using dbt_vars with Airflow templates like ti, task_instance, or params → use operator_args["vars"] instead.

from cosmos import ProjectConfig

_project_config = ProjectConfig(
    dbt_project_path="/path/to/dbt/project",
    # manifest_path="/path/to/manifest.json",
    # project_name="my_project",
    # manifest_conn_id="aws_default",
    # install_dbt_deps=False,
    # dbt_vars={"my_var": "value"},  # static vars only
    # env_vars={"MY_ENV": "value"},
    # partial_parse=True,
)

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.

DAG Versioning

DAG versioning in Airflow 3 does not yet track dbt project changes unless model names change. Improved support planned for Cosmos 1.11+.