python-data-engineer

---
name: python-data-engineer
description: Use when working with SQLAlchemy 2.0, Alembic migrations, database administration, data lineage visualization, query optimization, connection pooling, or database monitoring in Python. Covers PostgreSQL, MySQL, SQL Server administration patterns and data pipeline lineage tracking.
family: python
---

# Python Data Engineer

## Overview

Database-centric Python development: SQLAlchemy 2.0 ORM/Core, Alembic migrations, DBA automation, data lineage tracking, and query optimization. SQLAlchemy 2.0 was a major rewrite — use the new patterns (DeclarativeBase, mapped_column, unified query API).

## SQLAlchemy 2.0

### Core vs ORM Decision

| Use Case | Approach |
|----------|---------|
| Application data layer, business logic | ORM (mapped classes) |
| Raw performance, complex queries, reporting | Core (table objects, text()) |
| Bulk inserts/updates | Core (`insert().values([...])`) |
| Simple CRUD | ORM |

### Model Pattern (2.0 Style)

```python
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from sqlalchemy import String, ForeignKey

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = 'users'
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True)
    orders: Mapped[list["Order"]] = relationship(back_populates="user")
```

**Don't use:** `declarative_base()`, `Column()`, `relationship()` without `Mapped[]` — these are 1.x patterns.

### Session Management

```python
# Sync
from sqlalchemy.orm import Session
with Session(engine) as session:
    user = session.get(User, user_id)
    session.commit()

# Async
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
async_session = async_sessionmaker(async_engine)
async with async_session() as session:
    result = await session.execute(select(User).where(User.id == user_id))
```

**Flask:** Flask-SQLAlchemy handles scoped sessions — don't create your own.

### Query Loading Strategies

| Strategy | When | SQLAlchemy |
|----------|------|------------|
| **selectinload** | One-to-many, known small sets | `selectinload(User.orders)` |
| **joinedload** | One-to-one, always needed together | `joinedload(User.profile)` |
| **subqueryload** | Large collections, avoid N+1 | `subqueryload(User.orders)` |
| **raiseload** | Prevent lazy loading (catch N+1 in dev) | `raiseload('*')` |
| **lazyload** | Default, avoid in loops | Don't use in loops |

**Rule:** Always specify loading strategy for relationships in queries. Default lazy loading causes N+1 queries.

### Connection Pooling

```python
engine = create_engine(
    DATABASE_URL,
    pool_size=10,            # Steady-state connections
    max_overflow=20,         # Burst connections
    pool_recycle=300,        # Recycle every 5 min (prevent stale)
    pool_pre_ping=True,      # Verify before use
    pool_timeout=30,         # Wait max 30s for connection
)
```

## Alembic Migrations

```bash
alembic init alembic          # Initialize
alembic revision --autogenerate -m "add users table"  # Auto-detect changes
alembic upgrade head          # Apply all pending
alembic downgrade -1          # Rollback one
alembic history               # Show migration history
alembic current               # Show current revision
```

### Best Practices

- **Always review** autogenerated migrations — they miss: table renames (detected as drop+create), data migrations, index naming, constraints
- Test migrations on a copy of production data before deploying
- Never edit a migration that's been applied to shared environments
- Use `--sql` flag to preview SQL without executing
- For branching: `alembic merge heads` to resolve multiple heads

## Database Administration

### Health Monitoring Queries (PostgreSQL)

```python
# Active connections
session.execute(text("SELECT count(*) FROM pg_stat_activity WHERE state = 'active'"))

# Long-running queries (>5s)
session.execute(text("""
    SELECT pid, now() - pg_stat_activity.query_start AS duration, query
    FROM pg_stat_activity
    WHERE state != 'idle' AND now() - pg_stat_activity.query_start > interval '5 seconds'
"""))

# Table bloat / dead tuples
session.execute(text("SELECT relname, n_dead_tup FROM pg_stat_user_tables ORDER BY n_dead_tup DESC LIMIT 10"))

# Index usage
session.execute(text("""
    SELECT indexrelname, idx_scan, idx_tup_read
    FROM pg_stat_user_indexes ORDER BY idx_scan ASC LIMIT 20
"""))
```

### Automated Maintenance

| Task | Frequency | Method |
|------|-----------|--------|
| VACUUM ANALYZE | Daily (or autovacuum) | `VACUUM ANALYZE tablename` |
| REINDEX | Weekly for high-write tables | `REINDEX INDEX indexname` |
| pg_stat_statements reset | Weekly | `SELECT pg_stat_statements_reset()` |
| Connection pool stats | Continuous | Monitor pool.status() |
| Backup verification | Daily | Test restore of latest backup |

### Driver Comparison

| Database | Recommended Driver | Async Driver | Notes |
|----------|-------------------|--------------|-------|
| PostgreSQL | **psycopg 3.x** | **asyncpg** (5x faster) | psycopg3 also has async mode |
| MySQL/MariaDB | mysql-connector-python | aiomysql | mysqlclient fastest for sync |
| SQL Server | **mssql-python** (new, Nov 2025) | N/A | 40% faster than pyodbc for large results |
| SQLite | sqlite3 (stdlib) | aiosqlite | For dev/testing only |

## Data Lineage

### What It Tracks

Source → Transformation → Destination at table and column level. Critical for GDPR compliance (where does personal data flow?), debugging data issues (where did bad data enter?), and impact analysis (what breaks if I change this table?).

### Open-Source Lineage Stack

| Tool | Purpose | Python Integration |
|------|---------|-------------------|
| **OpenLineage** | Standard spec for lineage events | `openlineage-python` |
| **Marquez** | Lineage metadata store + API (reference impl) | REST API |
| **DataHub** | Data catalog + lineage (LinkedIn) | Python SDK |
| **OpenMetadata** | Catalog + lineage + quality | Python SDK |
| **dbt** | Built-in lineage graph | `dbt docs generate` |
| **SQLLineage** | Parse SQL to extract lineage | `sqllineage` |

### Custom Lineage Tracking

For custom ETL pipelines, emit OpenLineage events:

```python
from openlineage.client import OpenLineageClient
from openlineage.client.run import RunEvent, RunState, Job, Run, InputDataset, OutputDataset

client = OpenLineageClient(url="http://marquez:5000")
client.emit(RunEvent(
    eventType=RunState.COMPLETE,
    job=Job(namespace="etl", name="load_products"),
    run=Run(runId=str(uuid4())),
    inputs=[InputDataset(namespace="db2", name="MAINFRAME.PRODUCTS")],
    outputs=[OutputDataset(namespace="postgres", name="public.products")],
))
```

### Lineage Visualization

| Library | Best For |
|---------|----------|
| NetworkX + matplotlib | Quick static graphs |
| Graphviz (pygraphviz) | DAG rendering, export to SVG/PNG |
| D3.js / Dagre | Interactive web-based lineage graphs |
| Cytoscape.js | Complex network visualization |

## Database Security

| Rule | Implementation |
|------|---------------|
| Never build SQL with string formatting | Always use `$wpdb->prepare()` / parameterized queries / ORM |
| Connection strings out of code | Env vars, HashiCorp Vault, AWS Secrets Manager |
| Principle of least privilege | Service accounts with minimal permissions |
| TLS for database connections | `sslmode=require` (Postgres), `ssl=true` (MySQL) |
| Audit access | pgaudit (Postgres), general_log (MySQL) |
| Row-level security | PostgreSQL RLS policies for multi-tenant |
| Credential rotation | Vault dynamic secrets or Secrets Manager rotation |

## Anti-Patterns

| Don't | Why |
|-------|-----|
| Use SQLAlchemy 1.x patterns (Column, declarative_base) | 2.0 has better type safety and performance |
| Rely on lazy loading in loops | N+1 queries — specify loading strategy |
| Skip `pool_pre_ping` | Stale connections cause intermittent failures |
| Edit applied migrations | Breaks migration history — create new migration |
| Build SQL with f-strings or .format() | SQL injection — always parameterize |
| Store connection strings in code | Secrets exposure — use env vars or vault |
| Skip VACUUM on high-write PostgreSQL tables | Table bloat, degraded performance |
| Use SQLite in production | No concurrency, no replication — PostgreSQL minimum |
| Create engine per request | Connection overhead — use pooling |