Building a Modern Data Pipeline with Python and PostgreSQL

Python Stack: 3 Core Libraries

SQLAlchemy 2.0 with QueuePool and pool_pre_ping=True. Use COPY instead of INSERT for bulk loads — 10-100x faster:

engine = create_engine(
    "postgresql+psycopg2://user:pass@localhost/db",
    pool_size=10, max_overflow=20,
    pool_pre_ping=True,
)

# FOR UPDATE SKIP LOCKED — essential for parallel workers
def fetch_unprocessed(limit=1000):
    with engine.connect() as conn:
        return conn.execute(text("""
            SELECT id, payload FROM raw_events
            WHERE processed = FALSE
            LIMIT :limit
            FOR UPDATE SKIP LOCKED
        """), {"limit": limit}).fetchall()

Pandas / Polars: pd.read_sql() reads directly from PostgreSQL. Polars + ConnectorX is 5-20x faster than Pandas for large datasets. Write back using to_sql(..., method='multi', chunksize=5000).

Concurrency: 3 Patterns You Must Know

Pattern 1 — ThreadPoolExecutor for I/O-bound tasks (API calls, DB queries). max_workers=8 is the sweet spot for most pipelines.

Pattern 2 — asyncpg + asyncio for high concurrency (500+ concurrent operations):

pool = await asyncpg.create_pool("postgresql://...", min_size=5, max_size=20)

async def fetch_events(batch_size=500):
    async with pool.acquire() as conn:
        async with conn.transaction():
            return await conn.fetch("""
                SELECT id, event_type, payload FROM raw_events
                WHERE processed = FALSE LIMIT $1
                FOR UPDATE SKIP LOCKED
            """, batch_size)

Pattern 3 — FOR UPDATE SKIP LOCKED: When Worker 1 locks 100 rows, Worker 2 automatically skips them — no deadlocks, no duplicate processing. This is the single most important pattern for parallel workers.

Orchestration: Prefect 3 vs Airflow

Prefect: @flow + @task decorators, automatic retries, built-in UI. Covers 95% of startup pipeline needs out of the box.

Query Optimization: 5 Index Strategies

-- 1. Partial index — small, fast, only indexes what you need
CREATE INDEX idx_unprocessed ON raw_events (received_at)
  WHERE processed = FALSE;

-- 2. Composite index — order matters (equality first, range second)
CREATE INDEX idx_user_time ON fact_user_events (user_id, event_at DESC);

-- 3. GIN index for JSONB
CREATE INDEX idx_payload ON raw_events USING GIN (payload);

-- 4. Expression index for computed values
CREATE INDEX idx_user_id_str ON raw_events ((payload->>'user_id'));

-- 5. BRIN for time-series — tiny, efficient for append-only tables
CREATE INDEX idx_time_brin ON raw_events USING BRIN (received_at);

Use REFRESH MATERIALIZED VIEW CONCURRENTLY for heavy aggregations — no read lock during refresh.

Always debug with EXPLAIN (ANALYZE, BUFFERS). Watch for: Seq Scan = missing index; Row estimate way off = run ANALYZE; target cache hit ratio above 95%.

Dead Letter Queue — Never Lose Records

Failed records don't crash the entire batch — push to a DLQ table with error_msg, attempt_count, last_failed_at. Alert on sudden DLQ growth. Enable manual review and replay of failed records.

When to Use PostgreSQL (and When Not To)

Production Checklist

• Time-based partitioning for large tables
• Partial + GIN indexes where appropriate
• pool_pre_ping=True, PgBouncer for many connections
• FOR UPDATE SKIP LOCKED with parallel workers
• Dead letter queue with alerting on sudden growth
• Idempotent processing (re-running batches creates no duplicates)
• pg_stat_statements enabled, cache hit ratio >95%

PostgreSQL + Python is the most pragmatic stack for startup to mid-size companies. Add ClickHouse or Kafka-scale tools only when you outgrow it — don't over-engineer from the start.

Want to add AI processing to your data pipeline? See the AI agent builder's guide.