Loading...
Loading...
Use when turning a dbt Core project into an Airflow DAG/TaskGroup using Astronomer Cosmos. Does not cover dbt Fusion. Before implementing, verify dbt engine, warehouse, Airflow version, execution environment, DAG vs TaskGroup, and manifest availability.
npx skill4agent add astronomer/agents cosmos-dbt-coreVersion 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, (6) manifest availability.
| Load mode | When to use | Required inputs | Constraints |
|---|---|---|---|
| Large projects; containerized execution; fastest | | Remote manifest needs |
| Complex selectors; need dbt-native selection | dbt installed OR | Cannot use with containerized execution |
| dbt_ls selection without running dbt_ls every parse | | |
| Simple setups; let Cosmos pick | (none) | Falls back: manifest → dbt_ls → custom |
CRITICAL: Containerized execution (/DOCKER/etc.) → MUST useKUBERNETESload mode.dbt_manifest
from cosmos import RenderConfig, LoadMode
_render_config = RenderConfig(
load_method=LoadMode.DBT_MANIFEST, # or DBT_LS, DBT_LS_FILE, AUTOMATIC
)Reference: See reference/cosmos-config.md for detailed configuration examples per mode.
| Execution mode | When to use | Speed | Required setup |
|---|---|---|---|
| Fastest; single | Fastest | dbt adapter in env OR |
| dbt + adapter in Airflow env | Fast | dbt 1.5+ in |
| dbt in venv baked into image | Medium | |
| BigQuery + long-running transforms | Varies | Airflow ≥2.8; provider deps |
| Can't modify image; runtime venv | Slower | |
| Containerized | Full isolation per task | Slowest | manifest required; container config |
CRITICAL: Containerized execution (/DOCKER/etc.) → MUST useKUBERNETESload mode.dbt_manifest
from cosmos import ExecutionConfig, ExecutionMode
_execution_config = ExecutionConfig(
execution_mode=ExecutionMode.LOCAL, # or WATCHER, VIRTUALENV, AIRFLOW_ASYNC, KUBERNETES, etc.
)Reference: See reference/cosmos-config.md for detailed ProfileConfig options and all ProfileMapping classes.
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"},
),
)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",
)| Approach | When to use | Required param |
|---|---|---|
| Project path | Files available locally | |
| Manifest only | | |
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
)Reference: See reference/cosmos-config.md for detailed testing options.
| TestBehavior | Behavior |
|---|---|
| Tests run immediately after each model (default) |
| Combine run + test into single |
| All tests after all models complete |
| Skip tests |
from cosmos import RenderConfig, TestBehavior
_render_config = RenderConfig(
test_behavior=TestBehavior.AFTER_EACH,
)Reference: See reference/cosmos-config.md 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\') }}"}',
}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",
)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()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)| Airflow 3.x | Airflow 2.x |
|---|---|
| |
| |
postgres://0.0.0.0:5434/postgres.public.orderspostgres://0.0.0.0:5434/postgres/public/ordersCRITICAL: Update asset URIs when upgrading to Airflow 3.
AIRFLOW__COSMOS__ENABLE_MEMORY_OPTIMISED_IMPORTS=Truefrom cosmos.airflow.dag import DbtDag # instead of: from cosmos import DbtDagAIRFLOW__COSMOS__REMOTE_TARGET_PATH=s3://bucket/target_dir/
AIRFLOW__COSMOS__REMOTE_TARGET_PATH_CONN_ID=aws_defaultfrom cosmos.io import upload_to_cloud_storage
my_dag = DbtDag(
# ...
operator_args={"callback": upload_to_cloud_storage},
)AIRFLOW__COSMOS__DBT_DOCS_PROJECTS='{
"my_project": {
"dir": "s3://bucket/docs/",
"index": "index.html",
"conn_id": "aws_default",
"name": "My Project"
}
}'