astronomer_cosmos_dbtdag_airflow_postgres_quickstart.py

1import os
2from datetime import datetime
3from pathlib import Path
4
5from cosmos import DbtDag, ProjectConfig, ProfileConfig, ExecutionConfig
6from cosmos.profiles import PostgresUserPasswordProfileMapping
7
8# The path to the dbt project
9DEFAULT_DBT_ROOT_PATH = Path(__file__).parent / "dbt"
10DBT_ROOT_PATH = Path(os.getenv("DBT_ROOT_PATH", DEFAULT_DBT_ROOT_PATH))
11
12profile_config = ProfileConfig(
13    profile_name="default",
14    target_name="dev",
15    profile_mapping=PostgresUserPasswordProfileMapping(
16        conn_id="airflow_db",
17        profile_args={"schema": "public"},
18    ),
19)
20
21my_cosmos_dag = DbtDag(
22    project_config=ProjectConfig(
23        DBT_ROOT_PATH / "jaffle_shop",
24    ),
25    profile_config=profile_config,
26    execution_config=ExecutionConfig(
27        dbt_executable_path=f"{os.environ['AIRFLOW_HOME']}/dbt_venv/bin/dbt",
28    ),
29    # normal dag parameters
30    schedule_interval="@daily",
31    start_date=datetime(2023, 1, 1),
32    catchup=False,
33    dag_id="jaffle_shop",
34    default_args={"retries": 2},
35)