python-data-pipelines

Python Data Pipelines

Acknowledgement: Shared by Peter Bamuhigire, techguypeter.com, +256 784 464178.

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.
The task needs reusable judgment, domain constraints, or a proven workflow rather than ad hoc advice.

Do Not Use When

The task is unrelated to python-data-pipelines or would be better handled by a more specific companion skill.
The request only needs a trivial answer and none of this skill's constraints or references materially help.

Required Inputs

Gather relevant project context, constraints, and the concrete problem to solve; load references only as needed.
Confirm the desired deliverable: design, code, review, migration plan, audit, or documentation.

Workflow

Read this SKILL.md first, then load only the referenced deep-dive files that are necessary for the task.
Apply the ordered guidance, checklists, and decision rules in this skill instead of cherry-picking isolated snippets.
Produce the deliverable with assumptions, risks, and follow-up work made explicit when they matter.

Quality Standards

Keep outputs execution-oriented, concise, and aligned with the repository's baseline engineering standards.
Preserve compatibility with existing project conventions unless the skill explicitly requires a stronger standard.
Prefer deterministic, reviewable steps over vague advice or tool-specific magic.

Anti-Patterns

Treating examples as copy-paste truth without checking fit, constraints, or failure modes.
Loading every reference file by default instead of using progressive disclosure.

Outputs

A concrete result that fits the task: implementation guidance, review findings, architecture decisions, templates, or generated artifacts.
Clear assumptions, tradeoffs, or unresolved gaps when the task cannot be completed from available context alone.
References used, companion skills, or follow-up actions when they materially improve execution.

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:

Idempotent — same input twice = same end state. No duplicate records, no doubled side effects. Deduplicate by natural key or idempotency_key.
Resumable — interrupted midway, the next run picks up where it left off (via cursor, watermark, or checkpoint).
Observable — every run emits a start event, progress events, and a terminal event (success/failure/partial). Metrics for rows in/out, duration, lag.
Validated — inputs parsed through Pydantic at ingestion. Invalid records go to a dead-letter queue, not to /dev/null.
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:

Grayscale, then denoise.
Adaptive threshold (not global).
Deskew (hough lines).
Upscale 2×–3× for small fonts.
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.

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

About this skill

Use When

Do Not Use When

Required Inputs

Workflow

Quality Standards

Anti-Patterns

Outputs

Evidence Produced

References

When this skill applies

Pipeline architecture principles

The canonical pipeline shape

ETL / external API sync

Document intelligence — OCR

PDF extraction

Image / media processing

Scheduling

Validation & dead-letter queue

Multi-tenant pipelines

Observability for pipelines

Anti-patterns

Read next

References