MINOR: Add tutorial of dag with hooks s2 and postgresql

Author: coder2j

README.md

@@ -1,17 +1,18 @@
-# Apache Airflow  Tutorial Series [YouTube](https://www.youtube.com/watch?v=z7xyNOF8tak&list=PLwFJcsJ61oujAqYpMp1kdUBcPG0sE0QMT)
+# Apache Airflow Tutorial Series [YouTube](https://www.youtube.com/watch?v=z7xyNOF8tak&list=PLwFJcsJ61oujAqYpMp1kdUBcPG0sE0QMT)
 ## Updated Tutorial Episode
-1. [Introduction and Local Installation](https://youtu.be/z7xyNOF8tak)
-2. [Get Airflow running in Docker](https://youtu.be/J6azvFhndLg)
-3. [Airflow Core Concepts in 5 mins](https://youtu.be/mtJHMdoi_Gg)
-4. [Airflow Task Lifecycle and Basic Architecture](https://youtu.be/UFsCvWjQT4w)
-5. [Airflow DAG with BashOperator](https://youtu.be/CLkzXrjrFKg)
-6. [Airflow DAG with PythonOperator and XComs](https://youtu.be/IumQX-mm20Y)
-7. [Airflow TaskFlow API](https://youtu.be/9y0mqWsok_4)
-8. [Airflow Catchup and Backfill](https://youtu.be/OXOiUeHOQ-0)
-9. [Schedule Airflow DAG with Cron Expression](https://youtu.be/tpuovQFUByk)
-10. [Airflow Connection and PostgresOperator](https://youtu.be/S1eapG6gjLU)
-11. [Add Python Dependencies via Airflow Docker Image Extending and Customizing](https://youtu.be/0UepvC9X4HY)
-12. [AWS S3 Key Sensor Operator](https://youtu.be/vuxrhipJMCk)
+1. [Introduction and Local Installation](https://www.youtube.com/watch?v=z7xyNOF8tak&list=PLwFJcsJ61oujAqYpMp1kdUBcPG0sE0QMT&index=1)
+2. [Get Airflow running in Docker](https://www.youtube.com/watch?v=J6azvFhndLg&list=PLwFJcsJ61oujAqYpMp1kdUBcPG0sE0QMT&index=2)
+3. [Airflow Core Concepts in 5 mins](https://www.youtube.com/watch?v=mtJHMdoi_Gg&list=PLwFJcsJ61oujAqYpMp1kdUBcPG0sE0QMT&index=3)
+4. [Airflow Task Lifecycle and Basic Architecture](https://www.youtube.com/watch?v=UFsCvWjQT4w&list=PLwFJcsJ61oujAqYpMp1kdUBcPG0sE0QMT&index=4)
+5. [Airflow DAG with BashOperator](https://www.youtube.com/watch?v=CLkzXrjrFKg&list=PLwFJcsJ61oujAqYpMp1kdUBcPG0sE0QMT&index=5)
+6. [Airflow DAG with PythonOperator and XComs](https://www.youtube.com/watch?v=IumQX-mm20Y&list=PLwFJcsJ61oujAqYpMp1kdUBcPG0sE0QMT&index=6)
+7. [Airflow TaskFlow API](https://www.youtube.com/watch?v=9y0mqWsok_4&list=PLwFJcsJ61oujAqYpMp1kdUBcPG0sE0QMT&index=7)
+8. [Airflow Catchup and Backfill](https://www.youtube.com/watch?v=OXOiUeHOQ-0&list=PLwFJcsJ61oujAqYpMp1kdUBcPG0sE0QMT&index=8)
+9. [Schedule Airflow DAG with Cron Expression](https://www.youtube.com/watch?v=tpuovQFUByk&list=PLwFJcsJ61oujAqYpMp1kdUBcPG0sE0QMT&index=9)
+10. [Airflow Connection and PostgresOperator](https://www.youtube.com/watch?v=S1eapG6gjLU&list=PLwFJcsJ61oujAqYpMp1kdUBcPG0sE0QMT&index=10)
+11. [Add Python Dependencies via Airflow Docker Image Extending and Customizing](https://www.youtube.com/watch?v=0UepvC9X4HY&list=PLwFJcsJ61oujAqYpMp1kdUBcPG0sE0QMT&index=11)
+12. [AWS S3 Key Sensor Operator](https://www.youtube.com/watch?v=vuxrhipJMCk&list=PLwFJcsJ61oujAqYpMp1kdUBcPG0sE0QMT&index=12)
+13. [Airflow Hooks S3 PostgreSQL](https://www.youtube.com/watch?v=rcG4WNwi900&list=PLwFJcsJ61oujAqYpMp1kdUBcPG0sE0QMT&index=13)
 
 ## Running apache airflow 2.0 in docker with local executor.
 Here are the steps to take to get airflow 2.0 running with docker on your machine. 

dags/dag_with_postgres_hooks.py

@@ -0,0 +1,56 @@
+import csv
+import logging
+from datetime import datetime, timedelta
+from tempfile import NamedTemporaryFile
+
+from airflow import DAG
+from airflow.operators.python import PythonOperator
+from airflow.providers.postgres.hooks.postgres import PostgresHook
+from airflow.providers.amazon.aws.hooks.s3 import S3Hook
+
+
+default_args = {
+    'owner': 'coder2j',
+    'retries': 5,
+    'retry_delay': timedelta(minutes=10)
+}
+
+
+def postgres_to_s3(ds_nodash, next_ds_nodash):
+    # step 1: query data from postgresql db and save into text file
+    hook = PostgresHook(postgres_conn_id="postgres_localhost")
+    conn = hook.get_conn()
+    cursor = conn.cursor()
+    cursor.execute("select * from orders where date >= %s and date < %s",
+                   (ds_nodash, next_ds_nodash))
+    with NamedTemporaryFile(mode='w', suffix=f"{ds_nodash}") as f:
+    # with open(f"dags/get_orders_{ds_nodash}.txt", "w") as f:
+        csv_writer = csv.writer(f)
+        csv_writer.writerow([i[0] for i in cursor.description])
+        csv_writer.writerows(cursor)
+        f.flush()
+        cursor.close()
+        conn.close()
+        logging.info("Saved orders data in text file: %s", f"dags/get_orders_{ds_nodash}.txt")
+    # step 2: upload text file into S3
+        s3_hook = S3Hook(aws_conn_id="minio_conn")
+        s3_hook.load_file(
+            filename=f.name,
+            key=f"orders/{ds_nodash}.txt",
+            bucket_name="airflow",
+            replace=True
+        )
+        logging.info("Orders file %s has been pushed to S3!", f.name)
+
+
+with DAG(
+    dag_id="dag_with_postgres_hooks_v04",
+    default_args=default_args,
+    start_date=datetime(2022, 4, 30),
+    schedule_interval='@daily'
+) as dag:
+    task1 = PythonOperator(
+        task_id="postgres_to_s3",
+        python_callable=postgres_to_s3
+    )
+    task1