Initial version.

.gitignore

@@ -127,3 +127,7 @@ dmypy.json
 
 # Pyre type checker
 .pyre/
+
+# Ignore some Airflow folders
+dags/files/
+logs/

Procfile

@@ -0,0 +1,2 @@
+
+web:  airflow webserver --port $PORT --daemon & airflow scheduler

README.md

@@ -1,2 +1,6 @@
 # airflow
 Airflow Demo on Heroku
+
+## References
+
+- [Airflow Tutorial](https://airflow.apache.org/docs/apache-airflow/stable/tutorial.html)

airflow.cfg

@@ -0,0 +1,10 @@
+[core]
+sql_alchemy_conn = $DATABASE_URL
+
+# Do not load examples DAGs
+load_examples = False
+
+[webserver]
+authenticate = True
+auth_backend = airflow.contrib.auth.backends.password_auth
+rbac = True   # Required for role-based-access-control

app.json

@@ -0,0 +1,22 @@
+{
+    "name": "Airflow Demo on Heroku",
+    "description": "Airflow Demo on Heroku",
+    "repository": "https://github.com/slyapustin/celery-flower-heroku",
+    "keywords": [
+        "airflow",
+        "heroku",
+        "python"
+    ],
+    "addons": [
+        "heroku-postgresql:hobby-dev"
+    ],
+    "env": {
+        "AIRFLOW_HOME": {
+            "description": "Airflow Home Directory Location",
+            "value": "/app"
+        }
+    },
+    "scripts": {
+        "postdeploy": "airflow initdb"
+    }
+}

dags/etl.py

@@ -0,0 +1,79 @@
+import datetime
+import pendulum
+import os
+
+import requests
+from airflow.decorators import dag, task
+from airflow.providers.postgres.hooks.postgres import PostgresHook
+from airflow.providers.postgres.operators.postgres import PostgresOperator
+
+
+@dag(
+    #schedule_interval="0 0 * * *",
+    schedule_interval="*/5 * * * *",
+    start_date=pendulum.datetime(2022, 6, 19, tz="UTC"),
+    catchup=False,
+    dagrun_timeout=datetime.timedelta(minutes=60),
+)
+def Etl():
+    create_employees_table = PostgresOperator(
+        task_id="create_employees_table",
+        postgres_conn_id="tutorial_pg_conn",
+        sql="sql/employees_schema.sql",
+    )
+
+    create_employees_temp_table = PostgresOperator(
+        task_id="create_employees_temp_table",
+        postgres_conn_id="tutorial_pg_conn",
+        sql="sql/employees_temp_schema.sql",
+    )
+
+    @task
+    def get_data():
+        # NOTE: configure this as appropriate for your airflow environment
+        data_path = "/opt/airflow/dags/files/employees.csv"
+        os.makedirs(os.path.dirname(data_path), exist_ok=True)
+
+        url = "https://raw.githubusercontent.com/apache/airflow/main/docs/apache-airflow/pipeline_example.csv"
+
+        response = requests.request("GET", url)
+
+        with open(data_path, "w") as file:
+            file.write(response.text)
+
+        postgres_hook = PostgresHook(postgres_conn_id="tutorial_pg_conn")
+        conn = postgres_hook.get_conn()
+        cur = conn.cursor()
+        with open(data_path, "r") as file:
+            cur.copy_expert(
+                "COPY employees_temp FROM STDIN WITH CSV HEADER DELIMITER AS ',' QUOTE '\"'",
+                file,
+            )
+        conn.commit()
+
+    @task
+    def merge_data():
+        query = """
+            INSERT INTO employees
+            SELECT *
+            FROM (
+                SELECT DISTINCT *
+                FROM employees_temp
+            )
+            ON CONFLICT ("Serial Number") DO UPDATE
+            SET "Serial Number" = excluded."Serial Number";
+        """
+        try:
+            postgres_hook = PostgresHook(postgres_conn_id="tutorial_pg_conn")
+            conn = postgres_hook.get_conn()
+            cur = conn.cursor()
+            cur.execute(query)
+            conn.commit()
+            return 0
+        except Exception as e:
+            return 1
+
+    [create_employees_table, create_employees_temp_table] >> get_data() >> merge_data()
+
+
+dag = Etl()

dags/sql/employees_schema.sql

@@ -0,0 +1,7 @@
+CREATE TABLE IF NOT EXISTS employees (
+            "Serial Number" NUMERIC PRIMARY KEY,
+            "Company Name" TEXT,
+            "Employee Markme" TEXT,
+            "Description" TEXT,
+            "Leave" INTEGER
+        );

dags/sql/employees_temp_schema.sql

@@ -0,0 +1,8 @@
+DROP TABLE IF EXISTS employees_temp;
+        CREATE TABLE employees_temp (
+            "Serial Number" NUMERIC PRIMARY KEY,
+            "Company Name" TEXT,
+            "Employee Markme" TEXT,
+            "Description" TEXT,
+            "Leave" INTEGER
+        );

requirements.txt

@@ -0,0 +1 @@
+apache-airflow==2.3.2

runtime.txt

@@ -0,0 +1 @@
+python-3.9.13