MINOR: Add tutorial of airflow dag with postgres connection and operator

coder2j · coder2j · commit a843fc7e8012 · 2022-02-19T14:42:34.000+01:00
diff --git a/dags/dag_with_postgres_operator.py b/dags/dag_with_postgres_operator.py
@@ -0,0 +1,47 @@
+from datetime import datetime, timedelta
+
+from airflow import DAG
+from airflow.providers.postgres.operators.postgres import PostgresOperator
+
+
+default_args = {
+    'owner': 'coder2j',
+    'retries': 5,
+    'retry_delay': timedelta(minutes=5)
+}
+
+
+with DAG(
+    dag_id='dag_with_postgres_operator_v03',
+    default_args=default_args,
+    start_date=datetime(2021, 12, 19),
+    schedule_interval='0 0 * * *'
+) as dag:
+    task1 = PostgresOperator(
+        task_id='create_postgres_table',
+        postgres_conn_id='postgres_localhost',
+        sql="""
+            create table if not exists dag_runs (
+                dt date,
+                dag_id character varying,
+                primary key (dt, dag_id)
+            )
+        """
+    )
+
+    task2 = PostgresOperator(
+        task_id='insert_into_table',
+        postgres_conn_id='postgres_localhost',
+        sql="""
+            insert into dag_runs (dt, dag_id) values ('{{ ds }}', '{{ dag.dag_id }}')
+        """
+    )
+
+    task3 = PostgresOperator(
+        task_id='delete_data_from_table',
+        postgres_conn_id='postgres_localhost',
+        sql="""
+            delete from dag_runs where dt = '{{ ds }}' and dag_id = '{{ dag.dag_id }}';
+        """
+    )
+    task1 >> task3 >> task2
diff --git a/docker-compose.yaml b/docker-compose.yaml
@@ -66,6 +66,8 @@ services:
       POSTGRES_DB: airflow
     volumes:
       - postgres-db-volume:/var/lib/postgresql/data
+    ports:
+      - 5432:5432
     healthcheck:
       test: ["CMD", "pg_isready", "-U", "airflow"]
       interval: 5s
