# Python Data Engineering

> Implement data pipelines with SQLAlchemy 2.0 patterns (TypeDecorator, hybrid properties, events) and warehouse architectures (Kimball star schemas, medallion layers, SCD handling). Covers ETL/ELT design, dim_/fact_/stg_ conventions, and orchestration with dbt, Airflow, or Dagster. Reference for API-to-database flows or dimensional modeling with Python and PostgreSQL. Use when this capability is needed.

- Skill: `tomevault-io/python-data-engineering` (Agent Skill, multi-file: 2 files)
- Install (CLI): `npx skillmds@latest add tomevault-io/python-data-engineering`
- Raw SKILL.md: https://api.skillmd.com/api/skills/tomevault-io/python-data-engineering/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Data & Analytics
- Author: tomevault-io (https://skillmd.com/u/tomevault-io)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/tomevault-io/python-data-engineering

---


# Python Data Engineering

Comprehensive patterns for building production-grade data pipelines, dimensional models, and API integrations using SQLAlchemy 2.0+ and modern data warehousing practices.

## When to Use This Skill

Apply this skill when building **custom Python data pipelines** with SQLAlchemy and PostgreSQL.

**Use this skill for:**
- Building ETL/ELT pipelines with Python
- Transforming API responses to database models
- Designing dimensional warehouses (fact/dimension tables)
- Implementing schema evolution strategies
- Creating reusable data transformation layers
- Integrating external systems (SaaS APIs, ERPs) with PostgreSQL

**Consider frameworks FIRST for common SaaS integrations:**
- **Airbyte** (600+ connectors, open-source) - NetSuite, Salesforce, Stripe, etc.
- **Meltano/Singer** (300+ taps, code-centric) - Reproducible, Git-based pipelines
- **Fivetran** (500+ connectors, managed) - Fast time-to-market, higher cost
- **dbt** (SQL transformations only) - Post-load transformations, not extraction

**Build custom Python when:**
- No existing connector for your source system
- Complex business logic required during transformation
- Unique incremental sync requirements
- Strategic differentiator justifying maintenance cost
- Existing connectors don't support your customizations

### Build vs Buy Decision Tree

```
Need to integrate external data source?
│
├─ Check existing connectors (Airbyte, Fivetran, Meltano)
│  │
│  ├─ Connector exists + meets requirements
│  │  └─ [USE FRAMEWORK] - Faster, maintained, documented
│  │
│  └─ No connector OR custom logic needed
│     │
│     ├─ Simple transformation (SQL only)
│     │  └─ [USE dbt] - Post-load SQL transformations
│     │
│     └─ Complex transformation OR custom logic
│        └─ [BUILD CUSTOM PYTHON] - Full control, use patterns in this skill
│
└─ Consider:
   • Maintenance cost (custom code = ongoing maintenance)
   • Time to market (framework = faster initial setup)
   • Customization needs (custom code = unlimited flexibility)
   • Team expertise (SQL vs Python)
```

**Decision guide**: See [when-to-build-vs-buy.md](when-to-build-vs-buy.md) for detailed analysis

## Quick Reference by Topic

### SQLAlchemy Patterns
**TypeDecorators (custom column types)**: See [sqlalchemy/type-decorators.md](sqlalchemy/type-decorators.md)
**Factory methods (from_dict, from_api)**: See [sqlalchemy/factory-patterns.md](sqlalchemy/factory-patterns.md)
**Query patterns**: See [sqlalchemy/query-patterns.md](sqlalchemy/query-patterns.md)
**Repository pattern**: See [sqlalchemy/repository-pattern.md](sqlalchemy/repository-pattern.md)
**Migrations with Alembic**: See [sqlalchemy/migrations.md](sqlalchemy/migrations.md)

### Pydantic Integration
**API validation**: See [pydantic/api-validation.md](pydantic/api-validation.md)
**Settings management**: See [pydantic/settings-management.md](pydantic/settings-management.md)

### Data Warehouse Patterns
**Naming conventions (dim_, fact_)**: See [warehouse/naming-conventions.md](warehouse/naming-conventions.md)
**Slowly Changing Dimensions**: See [warehouse/scd-patterns.md](warehouse/scd-patterns.md)

### Integration Patterns
**Field mapping strategies**: See [integration/field-mapping.md](integration/field-mapping.md)
**Incremental sync (high-water mark)**: See [integration/incremental-sync.md](integration/incremental-sync.md)
**Schema resilience (JSONB)**: See [integration/schema-resilience.md](integration/schema-resilience.md)
**UV package management (local deps)**: See [integration/uv-package-management.md](integration/uv-package-management.md)

## Core Architecture Pattern

This skill teaches the **three-layer separation** for data pipelines:

```
Source System Layer (libs.external_api)
    └─ HTTP client, authentication, queries
    └─ Source-specific quirks and normalization

Database Schema Layer (libs.database)
    └─ Models (DimCustomer, FactOrders, etc.)
    └─ Mappers (API → Model transformations)
    └─ Custom types (TypeDecorators)

Application Layer (your applications)
    └─ Orchestration (when, what, how to sync)
    └─ Business logic (analysis, reporting)
    └─ CLI/API interfaces
```

**Why this separation:**
- Source layer changes when external API evolves
- Schema layer changes when business needs evolve
- Application layer changes when workflows evolve
- Clear boundaries prevent coupling

## Common Use Cases

### Use Case 1: API Response → Database Model

**Problem:** Transform external API response to SQLAlchemy model with type conversions

**Solution:** Factory classmethod + TypeDecorators
```python
class Customer(Base):
    email: Mapped[str] = mapped_column(NormalizedEmail)
    is_active: Mapped[bool] = mapped_column(BooleanStringType)

    @classmethod
    def from_api_response(cls, data: dict, sync_time: datetime) -> "Customer":
        # Transformation logic here
```

See [sqlalchemy/factory-patterns.md](sqlalchemy/factory-patterns.md) for complete pattern.

### Use Case 2: Custom Field Schema Evolution

**Problem:** SaaS system adds/removes custom fields, breaking rigid schema

**Solution:** JSONB column with lifecycle metadata
```python
class FlexibleRecordMixin:
    custom_fields: Mapped[dict] = mapped_column(JSONB)
    # Structure: {"field": {"value": ..., "first_seen": ..., "deprecated": bool}}
```

See [integration/schema-resilience.md](integration/schema-resilience.md) for complete pattern.

### Use Case 3: Incremental Sync from External System

**Problem:** Re-fetching all data is slow; need to sync only changed records

**Solution:** High-water mark pattern with sync metadata
```python
max_date = session.query(func.max(Customer.last_modified)).scalar()
api_response = client.get(f"/customers?modified_since={max_date}")
```

See [integration/incremental-sync.md](integration/incremental-sync.md) for complete pattern.

### Use Case 4: Multi-Source Data Integration

**Problem:** Combine data from multiple external systems into unified schema

**Solution:** Conformed dimensions + source-specific mappers
```python
class Customer(Base):
    source_system: Mapped[str]  # "api_a", "api_b", "internal"
    source_id: Mapped[str]       # Original system's ID

class APIACustomerMapper:
    @classmethod
    def to_customer(cls, data: dict) -> Customer:
        # API A schema → unified Customer schema

class APIBCustomerMapper:
    @classmethod
    def to_customer(cls, data: dict) -> Customer:
        # API B schema → unified Customer schema
```

Each mapper handles its source's quirks independently.

## Testing Your Data Pipeline

```python
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from datetime import datetime, UTC

@pytest.fixture
def db_session():
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)
    with Session(engine) as session:
        yield session

def test_api_to_model_transformation(db_session):
    api_response = {"customer_id": "C001", "name": "Test Corp", "email": "test@example.com"}
    customer = Customer.from_api_response(api_response, datetime.now(UTC))

    db_session.add(customer)
    db_session.commit()

    assert customer.customer_id == "C001"
    assert customer.name == "Test Corp"
```

## Best Practices Summary

### SQLAlchemy
- Use TypeDecorators for custom types (set `cache_ok=True`)
- Factory classmethods for API transformations
- Hybrid properties for computed columns
- Event listeners for validation/audit (not business logic)
- Repository pattern for data access abstraction

### Data Warehousing
- Star schema for analytics (fact tables + dimensions)
- Naming conventions: `dim_`, `fact_`, `stg_`, `bridge_`
- SCD Type 2 for historical tracking
- Medallion architecture: bronze (raw) → silver (clean) → gold (business)
- Audit columns: created_at, updated_at, synced_at, schema_version

### Integration
- Pydantic for API boundary validation
- Source-specific mappers for multi-system integration
- JSONB for flexible/evolving schemas
- Incremental sync with high-water marks
- Complete raw_data preservation for reprocessing
- Idempotent operations (safe to re-run)

---

*This skill provides practical patterns for Python data engineering with SQLAlchemy, Pydantic, and PostgreSQL.*

---
> Source: [AeyeOps/aeo-skill-marketplace](https://github.com/AeyeOps/aeo-skill-marketplace) — distributed by [TomeVault](https://tomevault.io).
<!-- tomevault:4.0:skill_md:2026-05-22 -->