PostgreSQL with Python and Airflow

Connecting from Python with psycopg2

import psycopg2

conn = psycopg2.connect(
    host="localhost",
    dbname="shop",
    user="postgres",
    password="mysecret"
)

cur = conn.cursor()
cur.execute("SELECT id, name, salary FROM employees WHERE department = %s", ("Engineering",))
rows = cur.fetchall()

for row in rows:
    print(row)

cur.close()
conn.close()

Always Use Parameterized Queries

Never build SQL with plain string formatting/concatenation — it opens the door to SQL injection.

# DANGEROUS — never do this:
department = "Engineering"
cur.execute(f"SELECT * FROM employees WHERE department = '{department}'")

# SAFE — let psycopg2 handle escaping:
cur.execute("SELECT * FROM employees WHERE department = %s", (department,))

Inserting Data and Committing

cur.execute(
    "INSERT INTO employees (name, email, salary) VALUES (%s, %s, %s)",
    ("Rahul Gupta", "rahul@gmail.com", 65000)
)
conn.commit()  # nothing is saved until you commit

Using pandas for Bulk Reads

import pandas as pd

df = pd.read_sql("SELECT * FROM employees", conn)
print(df.describe())

Orchestrating with Apache Airflow

Airflow schedules and monitors data pipelines as a DAG (directed acyclic graph) of tasks — a natural fit for the incremental load and reconciliation patterns from earlier lessons, run on a schedule.

from airflow import DAG
from airflow.providers.postgres.operators.postgres import PostgresOperator
from datetime import datetime

with DAG(
    dag_id="daily_sales_load",
    schedule="0 2 * * *",  # every day at 2 AM
    start_date=datetime(2024, 1, 1),
    catchup=False,
) as dag:

    load_incremental = PostgresOperator(
        task_id="load_incremental_orders",
        postgres_conn_id="warehouse_db",
        sql="""
            INSERT INTO warehouse.orders
            SELECT * FROM source.orders
            WHERE updated_at > (SELECT last_loaded_at FROM load_watermark WHERE source_name = 'orders');
        """,
    )

    update_watermark = PostgresOperator(
        task_id="update_watermark",
        postgres_conn_id="warehouse_db",
        sql="UPDATE load_watermark SET last_loaded_at = now() WHERE source_name = 'orders';",
    )

    load_incremental >> update_watermark

load_incremental >> update_watermark declares the dependency: the watermark only advances after the load succeeds.

Common Errors & Troubleshooting

• psycopg2.OperationalError: could not connect to server — same root causes as any Postgres connection issue: wrong host/port, the server isn't accepting connections, or a firewall is blocking it — verify with psql using the same credentials first.
• Changes not showing up in the database — a very common beginner bug: forgetting to call conn.commit() after INSERT/UPDATE/DELETE — without it, the changes stay in an open, uncommitted transaction and are invisible to other connections.
• Airflow DAG not appearing in the UI — check the DAG file has no syntax errors (Airflow silently skips broken DAG files in its scheduler logs) and that its dag_id is unique across your Airflow instance.

Key Takeaways

• Always pass query parameters separately (%s placeholders) rather than formatting them into the SQL string — this is the standard defense against SQL injection.
• Nothing written through psycopg2 persists until you call conn.commit().
• Airflow's PostgresOperator lets you schedule the exact SQL patterns from the Incremental Loads lesson as a reliable, monitored daily pipeline.