Fix k8s example to run with Cosmos 1.2

dags/jaffle_shop_kubernetes.py

@@ -2,22 +2,36 @@
 ## Jaffle Shop DAG
 [Jaffle Shop](https://github.com/dbt-labs/jaffle_shop) is a fictional eCommerce store. This dbt project originates from
 dbt labs as an example project with dummy data to demonstrate a working dbt core project. This DAG uses the cosmos dbt
-parser to generate an Airflow TaskGroup from the dbt project folder
+parser to generate an Airflow TaskGroup from the dbt project folder.
+
+
+The step-by-step to run this DAG are described in:
+https://astronomer.github.io/astronomer-cosmos/getting_started/kubernetes.html#kubernetes
 
 """
+from pathlib import Path
+
 from airflow import DAG
 from airflow.kubernetes.secret import Secret
 from pendulum import datetime
 
-from cosmos.providers.dbt.core.operators.kubernetes import DbtSeedKubernetesOperator, DbtRunKubernetesOperator
-from cosmos.providers.dbt.task_group import DbtTaskGroup
+from cosmos import (
+    ProjectConfig,
+    ProfileConfig,
+    ExecutionConfig,
+    ExecutionMode,
+    DbtSeedKubernetesOperator,
+    DbtRunKubernetesOperator,
+    DbtTaskGroup,
+)
+from cosmos.profiles import PostgresUserPasswordProfileMapping
 
-PROJECT_DIR = "dags/dbt/jaffle_shop"
-DBT_IMAGE = "dbt-jaffle-shop:latest"
+
+PROJECT_DIR = Path("dags/dbt/jaffle_shop/")
+DBT_IMAGE = "dbt-jaffle-shop:1.0.0"
 
 project_seeds = [
-        {"project": "jaffle_shop", "seeds": [
-            "raw_customers", "raw_payments", "raw_orders"]}
+    {"project": "jaffle_shop", "seeds": ["raw_customers", "raw_payments", "raw_orders"]}
 ]
 
 postgres_password_secret = Secret(
@@ -35,39 +49,43 @@
 )
 
 with DAG(
-    dag_id="jaffle_shop_k8s",
+    dag_id="jaffle_shop_kubernetes",
     start_date=datetime(2022, 11, 27),
     doc_md=__doc__,
     catchup=False,
 ) as dag:
-
     load_seeds = DbtSeedKubernetesOperator(
-        task_id=f"load_seeds",
+        task_id="load_seeds",
         project_dir=PROJECT_DIR,
         get_logs=True,
         schema="public",
         conn_id="postgres_default",
         image=DBT_IMAGE,
         is_delete_operator_pod=False,
-        secrets=[postgres_password_secret, postgres_host_secret]
+        secrets=[postgres_password_secret, postgres_host_secret],
     )
 
     run_models = DbtTaskGroup(
-        group_id="jaffle_shop",
-        dbt_project_name="jaffle_shop",
-        dbt_root_path="./dags/dbt/",
-        conn_id="postgres_default",
-        execution_mode="kubernetes",
+        profile_config=ProfileConfig(
+            profile_name="postgres_profile",
+            target_name="dev",
+            profile_mapping=PostgresUserPasswordProfileMapping(
+                conn_id="postgres_default",
+                profile_args={
+                    "schema": "public",
+                },
+            ),
+        ),
+        project_config=ProjectConfig(PROJECT_DIR),
+        execution_config=ExecutionConfig(
+            execution_mode=ExecutionMode.KUBERNETES,
+        ),
         operator_args={
             "image": DBT_IMAGE,
             "get_logs": True,
             "is_delete_operator_pod": False,
-            "secrets": [postgres_password_secret, postgres_host_secret]
+            "secrets": [postgres_password_secret, postgres_host_secret],
         },
-        dbt_args={
-            "schema": "public",
-            "project_dir": PROJECT_DIR,
-        }
     )
 
     load_seeds >> run_models