python-data-pipelines

name	python-data-pipelines
description	Use when building ETL jobs, document intelligence pipelines, OCR, PDF/Excel ingestion, image/media processing, or external-API sync pipelines in Python — idempotent scheduled jobs with validation, dead-letter queues, and multi-tenant isolation.
metadata	{"portable":true,"compatible_with":["Codex","codex"]}

Python Data Pipelines

Acknowledgement: Shared by Peter Bamuhigire, techguypeter.com, [phone].

Use When

• Use when building ETL jobs, document intelligence pipelines, OCR, PDF/Excel ingestion, image/media processing, or external-API sync pipelines in Python — idempotent scheduled jobs with validation, dead-letter queues, and multi-tenant isolation.

Evidence Produced

Category	Artifact	Format	Example
Data safety	ETL job spec	Markdown doc covering source contracts, idempotency, DLQ handling, and retention	docs/python/etl-spec-orders.md
Operability	Pipeline runbook	Markdown doc per skill-composition-standards/references/runbook-template.md covering schedule, retries, and DLQ inspection	docs/python/etl-runbook.md

References

• Use the references/ directory for deep detail after reading the core workflow below.

Bring data into the SaaS from outside systems and unstructured inputs — APIs, files, images, PDFs — and transform it for downstream use. Every pipeline here is idempotent, validated, and observable.

Prerequisites: Load python-modern-standards and python-saas-integration before this skill.

When this skill applies

• Syncing data from external APIs: Stripe, payment gateways, bank feeds, tax portals, government APIs.
• Extracting structured data from uploaded receipts, invoices, IDs (OCR).
• Parsing PDFs or Excel files uploaded by tenants.
• Processing uploaded images: resize, optimize, thumbnail, watermark.
• Scheduled jobs that transform data between stores.
• Reconciliation jobs that compare our records against external source-of-truth.

Pipeline architecture principles

Every pipeline enforces the same five properties:

1. Idempotent — same input twice = same end state. No duplicate records, no doubled side effects. Deduplicate by natural key or idempotency_key.
2. Resumable — interrupted midway, the next run picks up where it left off (via cursor, watermark, or checkpoint).
3. Observable — every run emits a start event, progress events, and a terminal event (success/failure/partial). Metrics for rows in/out, duration, lag.
4. Validated — inputs parsed through Pydantic at ingestion. Invalid records go to a dead-letter queue, not to /dev/null.
5. Multi-tenant-safe — every record carries tenant_id; every pipeline step validates it.

See references/pipeline-architecture.md.

The canonical pipeline shape

Source  ->  Extract  ->  Validate  ->  Transform  ->  Load  ->  Verify
               |                                                   ^
               v                                                   |
         Dead-letter  <----  validation failures                   |
                                                                   |
                             Reconciliation  <---------------------+

• Extract — pull data from the source (API, file, queue).
• Validate — Pydantic model per record. Fail fast on a malformed batch; per-record failures go to DLQ.
• Transform — map to domain shape. Decimal for money, UTC for timestamps, canonical enums.
• Load — upsert into MySQL (or destination). Transactional per tenant or per batch.
• Verify — count checks, hash checks, reconciliation against source totals.

ETL / external API sync

Typical pattern — nightly Stripe invoice sync:

from datetime import datetime, UTC
import structlog
from stripe import StripeClient
from sqlalchemy import text
from .checkpoints import load_watermark, save_watermark

log = structlog.get_logger()

def sync_stripe_invoices(tenant_id: int) -> SyncResult:
    watermark = load_watermark(tenant_id, "stripe.invoices") or datetime(1970, 1, 1, tzinfo=UTC)
    client = StripeClient(api_key=secrets.for_tenant(tenant_id).stripe_api_key)

    total, ok, failed = 0, 0, 0
    for page in client.invoices.list({"created": {"gte": int(watermark.timestamp())}, "limit": 100}):
        for raw in page.data:
            total += 1
            try:
                model = StripeInvoice.model_validate(raw)   # Pydantic, strict
                upsert_invoice(tenant_id, model)            # ON DUPLICATE KEY UPDATE
                ok += 1
            except ValidationError as e:
                send_to_dlq(tenant_id, "stripe.invoices", raw.id, str(e))
                failed += 1

    save_watermark(tenant_id, "stripe.invoices", datetime.now(UTC))
    log.info("stripe_sync_done", tenant_id=tenant_id, total=total, ok=ok, failed=failed)
    return SyncResult(total=total, ok=ok, failed=failed)

Key features:

• Watermark-based incremental — only fetches new data since last run.
• Upsert, not insert — reruns don't duplicate.
• Per-record DLQ — one bad record doesn't kill the batch.
• Per-tenant credentials — pulled from secrets vault by tenant.

Rate limits, pagination, auth refresh: all in references/etl-external-apis.md.

Document intelligence — OCR

Tesseract is the default. For machine-printed receipts and invoices it's accurate enough. For handwriting or low-quality scans, consider Google Vision / AWS Textract, but budget the cost.

import pytesseract
from PIL import Image, ImageOps
import cv2

def ocr_receipt(path: Path) -> str:
    img = cv2.imread(str(path))
    gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
    # Adaptive threshold for uneven lighting
    proc = cv2.adaptiveThreshold(gray, 255, cv2.ADAPTIVE_THRESH_GAUSSIAN_C, cv2.THRESH_BINARY, 11, 2)
    text = pytesseract.image_to_string(proc, lang="eng", config="--psm 6")
    return text

Preprocessing dominates accuracy:

1. Grayscale, then denoise.
2. Adaptive threshold (not global).
3. Deskew (hough lines).
4. Upscale 2×–3× for small fonts.
5. Remove borders/shadows.

See references/ocr-tesseract.md for multi-language, PSM modes, confidence extraction, and the decision table for when to use a cloud OCR service.

PDF extraction

Two tools, different jobs:

• pdfplumber — text PDFs (born-digital). Extracts text with position info, and tables when the layout is regular.
• camelot — tables in PDFs, better than pdfplumber for complex grids.
• ocrmypdf — wraps Tesseract to OCR scanned PDFs in place, producing a searchable PDF.

import pdfplumber

with pdfplumber.open(path) as pdf:
    for page in pdf.pages:
        text = page.extract_text()
        tables = page.extract_tables(table_settings={"vertical_strategy": "lines"})

Never trust uploaded PDFs — validate MIME type, size, and run through a scanner if doing anything beyond text extraction. See references/pdf-extraction.md. For the full library catalogue (PyMuPDF, pdfplumber, ebooklib, camelot, ocrmypdf, unstructured) plus system dependencies (poppler, tesseract, ghostscript), see references/document-parsing-toolkit.md.

Image / media processing

Pillow for most work. For heavy resizing / format conversion, add pillow-simd for a speed bump.

from PIL import Image, ImageOps

def make_thumbnail(src: Path, dst: Path, size: tuple[int, int] = (400, 400)) -> None:
    with Image.open(src) as img:
        img = ImageOps.exif_transpose(img)   # honor rotation EXIF
        img.thumbnail(size, Image.LANCZOS)
        img = img.convert("RGB")
        img.save(dst, "WEBP", quality=85, method=6)

Always strip EXIF on user uploads before serving publicly — GPS coordinates leak. Use Image.open(src); data = list(img.getdata()); new = Image.new(img.mode, img.size); new.putdata(data) or piexif.remove.

Watermarking: render over a corner; scale watermark to image width. See references/image-processing-pillow.md.

Scheduling

Three options:

• cron + uv run script.py — simplest. Great for once-daily jobs on a single host.
• APScheduler — in-process scheduler. Good when you want Python-managed cron-ish behavior without system cron.
• Celery beat — if you already run Celery for workers, use beat for schedules.
• RQ Scheduler (rq-scheduler) — if you use RQ.

Rule: use the simplest one that meets the need. Don't adopt Celery for one nightly job.

Always make scheduled jobs:

• Mutex against themselves (Redis lock with TTL) — long jobs can't overlap.
• Idempotent — safe to rerun.
• Tolerant of skipped runs — if Monday's job missed, Tuesday's should still do the right thing.

See references/scheduling.md.

Validation & dead-letter queue

Pydantic at the boundary, always:

class StripeInvoice(BaseModel):
    id: str = Field(..., pattern=r"^in_[a-zA-Z0-9]+$")
    customer: str
    amount_due: int = Field(..., ge=0)
    currency: str = Field(..., pattern=r"^[a-z]{3}$")
    status: Literal["draft", "open", "paid", "uncollectible", "void"]
    created: int
    model_config = {"extra": "ignore"}   # tolerate new fields from the API

DLQ design:

• Same Redis/DB as main queue, different namespace.
• Record: {tenant_id, pipeline, source_id, payload, error, first_seen, retry_count}.
• Human replay tool: mark fixed → re-enqueue into main flow.
• Alert when DLQ depth > threshold per tenant.

See references/validation-and-deadletter.md.

Multi-tenant pipelines

• Per-tenant credentials — every external source may have different API keys/tokens per tenant. Store in a vault (Hashicorp Vault, AWS Secrets Manager, or encrypted in DB) keyed by (tenant_id, integration_name).
• Rate limit per tenant — one noisy tenant shouldn't starve others. Use a token-bucket keyed by tenant_id.
• Concurrency budget per tenant — cap at N parallel jobs per tenant.
• Isolation in logs — log tenant_id on every line. Never log another tenant's data in a given tenant's job context.

See references/multi-tenant-pipelines.md.

Observability for pipelines

Every pipeline run should answer these questions from a dashboard:

• When did it last run?
• Did it succeed, partially succeed, or fail?
• How many records in / out / DLQ?
• How long did it take?
• What's the lag between source and destination (freshness)?

Standard metrics (Prometheus):

• pipeline_run_total{pipeline, tenant, status}
• pipeline_records_total{pipeline, tenant, outcome} (outcome = ok | failed | skipped)
• pipeline_duration_seconds{pipeline, tenant} (histogram)
• pipeline_lag_seconds{pipeline, tenant} (gauge — how stale is the destination)

Alerts:

• No successful run in > 2x expected interval.
• DLQ growth rate > threshold.
• Lag > SLA.

See references/observability-pipelines.md.

Anti-patterns

• Full reloads instead of incremental — wastes time, hammers the source. Use watermarks unless the source is tiny.
• One giant transaction for a 100K-row batch — lock contention + all-or-nothing rollback. Batch in 1,000–10,000 row chunks, commit between.
• Swallowing exceptions per record without DLQ — silent data loss. Always route failures somewhere.
• Scheduling in app code with while True: time.sleep(3600) — use a real scheduler.
• Storing external raw responses and the transformed record in the same row — mix of concerns. Store raw in a staging/audit table, transformed in the canonical table.
• Using the same Stripe client across tenants — cross-tenant credential bleed. One client per tenant.
• No rollback plan for bad sync data — build a "quarantine last N hours" path before you need it.

Read next

• python-data-analytics — to analyze the data once loaded.
• python-document-generation — to produce reports from pipeline outputs.
• photo-management — for the web/mobile upload side of image pipelines.

References

• references/pipeline-architecture.md
• references/etl-external-apis.md
• references/ocr-tesseract.md
• references/pdf-extraction.md
• references/document-parsing-toolkit.md — full catalogue of PDF + EPUB Python libraries (PyMuPDF, pdfplumber, pypdf, camelot, ocrmypdf, pdf2image, ebooklib, unstructured) plus the poppler / tesseract / ghostscript system dependencies
• references/image-processing-pillow.md
• references/scheduling.md
• references/validation-and-deadletter.md
• references/multi-tenant-pipelines.md
• references/observability-pipelines.md

Decision Rules

Condition	Action
Source can resend records	Use stable keys and idempotent upserts
Record is malformed but non-blocking	Quarantine it with reason and lineage
Schema or tenant identity is ambiguous	Stop before downstream writes

Capability Contract

Read and search are required. Editing, execution, API access, and data mutation require authorisation and least-privilege credentials.

Degraded Mode

Fallback: without source access or execution, produce the pipeline contract, validation rules, and replay plan; mark throughput and accuracy unverified.

Inputs

Artefact	Required?	Purpose
Source/target contracts, volumes, freshness, quality, and recovery requirements	yes	Design pipeline semantics

Outputs

• Produce pipeline design or code with schemas, idempotency, quality gates, observability, and recovery evidence.