Python Data Engineer

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

Model Pattern (2.0 Style)

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

# 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

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

Connection Pooling

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

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)

# 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

Driver Comparison

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

Custom Lineage Tracking

For custom ETL pipelines, emit OpenLineage events:

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

Database Security

Anti-Patterns