$ show yuelinou999/idempotent-pipeline-demo
This repository is a FastAPI + async SQLAlchemy reference implementation of an idempotent enterprise integration pipeline, explicitly labeled demo-grade rather than a production platform. Operators interact through REST endpoints covering idempotent message ingestion (five-branch deduplication: accept_new, accept_supersede, skip_duplicate, skip_stale, quarantine), simulated downstream delivery with retry classification, per-system circuit breakers, dead-letter queues, ambiguous-timeout handling with query-before-retry, audited operator replay, end-to-end record inspection, a dashboard health overview, and materialization/compaction jobs. Data flows from an ingestion write path through staging and production projections, then into delivery attempt logs, DLQs, ambiguous outcome queues, replay requests, and audit tables, with SQLite as the default database and PostgreSQL supported via a single environment variable change. The architecture is cleanly layered into routes, services, models, and schemas, with explicit bounded contexts for ingestion correctness, delivery resilience, and explainability. Seeded mock downstream systems make all delivery outcomes deterministic, which is appropriate for a demo harness. The codebase demonstrates good engineering practices: pure decision functions separated from DB side effects, flush-but-don't-commit transaction boundaries, append-only audit logs, and a comprehensive pytest suite covering scenario-based integration and unit tests. Notable caveats include placeholder rule-based correlation matching, some dead/duplicated schema definitions, a version mismatch between pyproject.toml (0.1.0) and the FastAPI app (0.7.0), and an unusual pyproject build-backend string that may break pip package builds. The repository appears maintained as of September 2026 with five open issues and modern dependency ranges. Technical drift is low: SQLAlchemy 2.0 async mappings, Pydantic v2, and current FastAPI patterns are used. While the application is not production-ready itself, several of its pure service modules are genuinely reusable building blocks for idempotency, retry, and correlation problems in other integration systems.
The LLM identified 5 files worth highlighting for reuse:
Implements the five-branch idempotency decision (accept_new, accept_supersede, skip_duplicate, quarantine, skip_stale) as a pure function returning a frozen Decision dataclass. The same-version/different-hash branch quarantines the record instead of silently overwriting it. Easily extractable as a standalone idempotency policy engine with no DB side effects.
Pure classifier mapping a delivery attempt (HTTP status, timeout flag, attempt number, retry ceiling) to retry_with_backoff, send_to_dlq, or send_to_ambiguous_queue. Explicit decision priority ensures timeouts and non-retryable errors are never misrouted. Reusable in any retry/DLQ orchestration layer.
Computes a SHA-256 fingerprint over canonical key-sorted JSON so semantically identical payloads with different key ordering hash the same. Stdlib-only, side-effect-free, and directly reusable as a dedup fingerprint helper.
Derives two deterministic keys: a version-less entity_key and a versioned idempotent_key. Handles unsafe business keys via base64 encoding and over-length keys via SHA-256 truncation, preventing delimiter collisions and unbounded key sizes. A clean, extractable pattern for distributed idempotency key design.
Resolves business correlation IDs across source systems with a priority chain: existing mapping, explicit reference, rule-based match, then self-generated fallback. Uses a DB UniqueConstraint plus nested-savepoint IntegrityError recovery to safely handle concurrent duplicate mappings. Adaptable to multi-source event correlation problems.