Cosmos Dbt Core

Cosmos + dbt Core: Implementation Checklist

Execute steps in order. Prefer the simplest configuration that meets the user's constraints.

Version note: This skill targets Cosmos 1.11+ and Airflow 3.x. If the user is on Airflow 2.x, adjust imports accordingly (see Appendix A).

>

Reference: Latest stable: https://pypi.org/project/astronomer-cosmos/

Before starting, confirm: (1) dbt engine = Core (not Fusion → use cosmos-dbt-fusion), (2) warehouse type, (3) Airflow version, (4) execution environment (Airflow env / venv / container), (5) DbtDag vs DbtTaskGroup vs individual operators, (6) manifest availability.

---

1. Configure Project (ProjectConfig)

from cosmos import ProjectConfig

_project_config = ProjectConfig(
    dbt_project_path="/path/to/dbt/project",
    # manifest_path="/path/to/manifest.json",  # for dbt_manifest load mode
    # project_name="my_project",  # if using manifest_path without dbt_project_path
    # install_dbt_deps=False,  # if deps precomputed in CI
)

2. Choose Parsing Strategy (RenderConfig)

Pick ONE load mode based on constraints:

CRITICAL: Containerized execution (DOCKER/KUBERNETES/etc.)

from cosmos import RenderConfig, LoadMode

_render_config = RenderConfig(
    load_method=LoadMode.DBT_MANIFEST,  # or DBT_LS, DBT_LS_FILE, AUTOMATIC
)

---

3. Choose Execution Mode (ExecutionConfig)

Reference: See [reference/cosmos-config.md](reference/cosmos-config.md#execution-modes-executionconfig) for detailed configuration examples per mode.

Pick ONE execution mode:

from cosmos import ExecutionConfig, ExecutionMode

_execution_config = ExecutionConfig(
    execution_mode=ExecutionMode.WATCHER,  # or LOCAL, VIRTUALENV, AIRFLOW_ASYNC, KUBERNETES, etc.
)

---

4. Configure Warehouse Connection (ProfileConfig)

Reference: See [reference/cosmos-config.md](reference/cosmos-config.md#profileconfig-warehouse-connection) for detailed ProfileConfig options and all ProfileMapping classes.

Option A: Airflow Connection + ProfileMapping (Recommended)

from cosmos import ProfileConfig
from cosmos.profiles import SnowflakeUserPasswordProfileMapping

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

Option B: Existing profiles.yml

CRITICAL: Do not hardcode secrets; use environment variables.

from cosmos import ProfileConfig

_profile_config = ProfileConfig(
    profile_name="my_profile",
    target_name="dev",
    profiles_yml_filepath="/path/to/profiles.yml",
)

---

5. Configure Testing Behavior (RenderConfig)

Reference: See [reference/cosmos-config.md](reference/cosmos-config.md#testing-behavior-renderconfig) for detailed testing options.

from cosmos import RenderConfig, TestBehavior

_render_config = RenderConfig(
    test_behavior=TestBehavior.AFTER_EACH,
)

---

6. Configure operator_args

Reference: See [reference/cosmos-config.md](reference/cosmos-config.md#operator_args-configuration) for detailed operator_args options.

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

    # Cosmos-specific params
    "install_deps": False,
    "full_refresh": False,
    "quiet": True,

    # Runtime dbt vars (XCom / params)
    "vars": '{"my_var": "{{ ti.xcom_pull(task_ids=\'pre_dbt\') }}"}',
}

---

7. Assemble DAG / TaskGroup

Option A: DbtDag (Standalone)

from cosmos import DbtDag, ProjectConfig, ProfileConfig, ExecutionConfig, RenderConfig
from cosmos.profiles import SnowflakeUserPasswordProfileMapping
from pendulum import datetime

_project_config = ProjectConfig(
    dbt_project_path="/usr/local/airflow/dbt/my_project",
)

_profile_config = ProfileConfig(
    profile_name="default",
    target_name="dev",
    profile_mapping=SnowflakeUserPasswordProfileMapping(
        conn_id="snowflake_default",
    ),
)

_execution_config = ExecutionConfig()
_render_config = RenderConfig()

my_cosmos_dag = DbtDag(
    dag_id="my_cosmos_dag",
    project_config=_project_config,
    profile_config=_profile_config,
    execution_config=_execution_config,
    render_config=_render_config,
    operator_args={},
    start_date=datetime(2025, 1, 1),
    schedule="@daily",
)

Option B: DbtTaskGroup (Inside Existing DAG)

from airflow.sdk import dag, task  # Airflow 3.x
# from airflow.decorators import dag, task  # Airflow 2.x
from airflow.models.baseoperator import chain
from cosmos import DbtTaskGroup, ProjectConfig, ProfileConfig, ExecutionConfig, RenderConfig
from pendulum import datetime

_project_config = ProjectConfig(dbt_project_path="/usr/local/airflow/dbt/my_project")
_profile_config = ProfileConfig(profile_name="default", target_name="dev")
_execution_config = ExecutionConfig()
_render_config = RenderConfig()

@dag(start_date=datetime(2025, 1, 1), schedule="@daily")
def my_dag():
    @task
    def pre_dbt():
        return "some_value"

    dbt = DbtTaskGroup(
        group_id="dbt_project",
        project_config=_project_config,
        profile_config=_profile_config,
        execution_config=_execution_config,
        render_config=_render_config,
    )

    @task
    def post_dbt():
        pass

    chain(pre_dbt(), dbt, post_dbt())

my_dag()

Option C: Use Cosmos operators directly

import os
from datetime import datetime
from pathlib import Path
from typing import Any

from airflow import DAG

try:
    from airflow.providers.standard.operators.python import PythonOperator
except ImportError:
    from airflow.operators.python import PythonOperator

from cosmos import DbtCloneLocalOperator, DbtRunLocalOperator, DbtSeedLocalOperator, ProfileConfig
from cosmos.io import upload_to_aws_s3

DEFAULT_DBT_ROOT_PATH = Path(__file__).parent / "dbt"
DBT_ROOT_PATH = Path(os.getenv("DBT_ROOT_PATH", DEFAULT_DBT_ROOT_PATH))
DBT_PROJ_DIR = DBT_ROOT_PATH / "jaffle_shop"
DBT_PROFILE_PATH = DBT_PROJ_DIR / "profiles.yml"
DBT_ARTIFACT = DBT_PROJ_DIR / "target"

profile_config = ProfileConfig(
    profile_name="default",
    target_name="dev",
    profiles_yml_filepath=DBT_PROFILE_PATH,
)


def check_s3_file(bucket_name: str, file_key: str, aws_conn_id: str = "aws_default", **context: Any) -> bool:
    """Check if a file exists in the given S3 bucket."""
    from airflow.providers.amazon.aws.hooks.s3 import S3Hook

    s3_key = f"{context['dag'].dag_id}/{context['run_id']}/seed/0/{file_key}"
    print(f"Checking if file {s3_key} exists in S3 bucket...")
    hook = S3Hook(aws_conn_id=aws_conn_id)
    return hook.check_for_key(key=s3_key, bucket_name=bucket_name)


with DAG("example_operators", start_date=datetime(2024, 1, 1), catchup=False) as dag:
    seed_operator = DbtSeedLocalOperator(
        profile_config=profile_config,
        project_dir=DBT_PROJ_DIR,
        task_id="seed",
        dbt_cmd_flags=["--select", "raw_customers"],
        install_deps=True,
        append_env=True,
    )

    check_file_uploaded_task = PythonOperator(
        task_id="check_file_uploaded_task",
        python_callable=check_s3_file,
        op_kwargs={
            "aws_conn_id": "aws_s3_conn",
            "bucket_name": "cosmos-artifacts-upload",
            "file_key": "target/run_results.json",
        },
    )

    run_operator = DbtRunLocalOperator(
        profile_config=profile_config,
        project_dir=DBT_PROJ_DIR,
        task_id="run",
        dbt_cmd_flags=["--models", "stg_customers"],
        install_deps=True,
        append_env=True,
    )

    clone_operator = DbtCloneLocalOperator(
        profile_config=profile_config,
        project_dir=DBT_PROJ_DIR,
        task_id="clone",
        dbt_cmd_flags=["--models", "stg_customers", "--state", DBT_ARTIFACT],
        install_deps=True,
        append_env=True,
    )

    seed_operator >> run_operator >> clone_operator
    seed_operator >> check_file_uploaded_task

Setting Dependencies on Individual Cosmos Tasks

from cosmos import DbtDag, DbtResourceType
from airflow.sdk import task, chain

with DbtDag(...) as dag:
    @task
    def upstream_task():
        pass

    _upstream = upstream_task()

    for unique_id, dbt_node in dag.dbt_graph.filtered_nodes.items():
        if dbt_node.resource_type == DbtResourceType.SEED:
            my_dbt_task = dag.tasks_map[unique_id]
            chain(_upstream, my_dbt_task)

---

8. Safety Checks

Before finalizing, verify:

• [ ] Execution mode matches constraints (AIRFLOW_ASYNC → BigQuery only)
• [ ] Warehouse adapter installed for chosen execution mode
• [ ] Secrets via Airflow connections or env vars, NOT plaintext
• [ ] Load mode matches execution (complex selectors → dbt_ls)
• [ ] Airflow 3 asset URIs if downstream DAGs scheduled on Cosmos assets (see Appendix A)

---

Appendix A: 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: Update asset URIs when upgrading to Airflow 3.

---

Appendix B: Operational Extras

Caching

Cosmos caches artifacts to speed up parsing. Enabled by default.

Reference: https://astronomer.github.io/astronomer-cosmos/configuration/caching.html

Memory-Optimized Imports

AIRFLOW__COSMOS__ENABLE_MEMORY_OPTIMISED_IMPORTS=True

When enabled:

from cosmos.airflow.dag import DbtDag  # instead of: from cosmos import DbtDag

Artifact Upload to Object Storage

AIRFLOW__COSMOS__REMOTE_TARGET_PATH=s3://bucket/target_dir/
AIRFLOW__COSMOS__REMOTE_TARGET_PATH_CONN_ID=aws_default

from cosmos.io import upload_to_cloud_storage

my_dag = DbtDag(
    # ...
    operator_args={"callback": upload_to_cloud_storage},
)

dbt Docs Hosting (Airflow 3.1+ / Cosmos 1.11+)

AIRFLOW__COSMOS__DBT_DOCS_PROJECTS='{
    "my_project": {
        "dir": "s3://bucket/docs/",
        "index": "index.html",
        "conn_id": "aws_default",
        "name": "My Project"
    }
}'

Reference: https://astronomer.github.io/astronomer-cosmos/configuration/hosting-docs.html

---

Related Skills

• cosmos-dbt-fusion: For dbt Fusion projects (not dbt Core)
• authoring-dags: General DAG authoring patterns
• testing-dags: Testing DAGs after creation

Cosmos Configuration Reference (Core)

This reference covers detailed Cosmos configuration for dbt Core projects.

Table of Contents

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

--

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. Those cannot be set via ProjectConfig because it is used during DAG parsing.

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,
    # copy_dbt_packages=False,
    # dbt_vars={"my_var": "value"},  # static vars only
    # env_vars={"MY_ENV": "value"},
    # partial_parse=True,
    # models_relative_path="custom_models_path",
    # seeds_relative_path="custom_seeds_path",
    # snapshots_relative_path="custom_snapshots_path",
)

---

Execution Modes (ExecutionConfig)

WATCHER Mode (Experimental, Fastest)

Known limitations:

• Implements DbtSeedWatcherOperator, DbtSnapshotWatcherOperator and DbtRunWatcherOperator - not other operators
• Built on top of ExecutionMode.LOCAL and ExecutionMode.KUBERNETES - not available for other execution modes
• Tests with TestBehavior.AFTER_EACH, which is the default test behavior, are still being rendered as EmptyOperators.
• May not work as expected when using RenderConfig.node_converters
• Airflow assets or datasets are emitted by the DbtProducerWatcherOperator instead by the actual tasks related to the correspondent dbt models.

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(
    # ...
    on_warning_callback=warning_callback,
)

---

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"

---

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+.