Hospital Bulk Processor

Fault-tolerant hospital CSV synchronization built with FastAPI, PostgreSQL, Redis, Celery, SQLAlchemy, and Alembic. The original /hospitals/bulk integration remains available; the durable pipeline is exposed under /api/v1/imports.

Architected for million-row imports. The repository includes a deterministic 1,000,000-row generator and manual benchmark workflow, but no million-row performance claim is made until that workflow has completed and its result is committed.

Architecture

flowchart LR
    A[Client upload] --> B[FastAPI]
    B --> C[Persistent CSV]
    B --> D[(PostgreSQL job)]
    D --> E[Orchestrator]
    E --> F[Durable JSONL chunks]
    E --> G[(Redis)]
    G --> W1[Celery worker 1]
    G --> W2[Celery worker 2]
    G --> WN[Celery worker N]
    W1 --> H[(Hospitals)]
    W2 --> H
    WN --> H
    W1 --> I[(Import failures)]
    J[Celery beat] --> K[Stale chunk recovery]
    K --> G

Loading

The upload is copied in 1 MiB blocks and never materialized in memory. The orchestration worker validates headers and iterates csv.DictReader one row at a time. Every configurable group (10,000 rows by default) is written to a durable JSONL staging file and represented by an import_chunks row. Redis messages contain only job and chunk IDs.

Workers atomically change a pending chunk to processing. PostgreSQL uses a conditional UPDATE ... RETURNING; this prevents duplicate delivery from claiming the same chunk. Workers normalize and validate rows, fetch existing keys once per database batch, and commit hospitals, failures, progress increments, chunk completion, and checkpoint advancement in one transaction.

Lifecycle and recovery

Jobs move through queued → parsing → processing → completed|completed_with_failures. Invalid headers, unreadable/lost source files, and exhausted chunk failures lead to failed; cancelled is reserved for cancellation support. Chunks use pending → processing → completed, with exhausted transient failures becoming dead_lettered.

Celery uses late acknowledgements, worker-lost rejection, and a prefetch multiplier of one. Database/IO/transient processing exceptions use exponential backoff with jitter, capped by MAX_RETRY_ATTEMPTS. Permanent row validation errors are persisted and do not stop unrelated records. Celery beat resets claims whose heartbeat is older than CHUNK_STALE_TIMEOUT_SECONDS and requeues only unfinished chunks.

Idempotency is enforced by the unique (source_name, external_hospital_id) hospital key and the unique chunk identity (job_id, chunk_number). Incoming source_updated_at values update only older destination rows; equal versions are unchanged and older versions are stale. The synchronization checkpoint advances inside the same transaction as successful chunk writes.

API

POST /api/v1/imports
GET  /api/v1/imports/{job_id}
GET  /api/v1/imports/{job_id}/failures?page=1&page_size=100
GET  /api/v1/imports/{job_id}/verification
POST /api/v1/imports/{job_id}/retry-failures
GET  /metrics

POST accepts multipart field file and optional form/query source_name, persists it, creates a job, enqueues orchestration, and returns 202:

{"job_id":"uuid","status":"queued","status_url":"/api/v1/imports/uuid"}

The legacy POST /hospitals/bulk, GET /jobs/{batch_id}, and GET /jobs behavior is preserved for compatibility. Those legacy jobs remain in-memory and retain their original 20-row limit; new integrations should use the durable API.

Local development

docker compose up --build
curl -F file=@hospitals.csv http://localhost:8000/api/v1/imports
docker compose run --rm api pytest

Compose starts the API, PostgreSQL 16, Redis 7, an eight-process Celery worker, and Celery beat. Uploaded files and database data use named volumes.

Without Docker, create a virtual environment, install requirements.txt, set DATABASE_URL, REDIS_URL, and IMPORT_STORAGE_PATH, run alembic upgrade head, then start Uvicorn and Celery separately.

Quality and tests

ruff check .
ruff format --check .
mypy app benchmark
pytest --cov=app --cov-report=term-missing
alembic upgrade head

GitHub Actions provisions real PostgreSQL and Redis services, validates migration upgrade/downgrade, runs lint/format/type/test/coverage checks, and processes a 25,000-row smoke dataset through a real API and Celery worker. The current coverage gate is 80%; the follow-up target is 85% after expanding failure-injection coverage. Eager Celery mode is available only for focused tests through CELERY_TASK_ALWAYS_EAGER=true; normal deployments use Redis and workers.

Benchmarks

python benchmark/generate_dataset.py --rows 1000000 --output benchmark/data/hospitals_1m.csv --invalid-rate 0.01
python benchmark/run_benchmark.py --file benchmark/data/hospitals_1m.csv --wait --verify

The generator streams deterministic data. The runner uploads the file, polls to a terminal state, and prints counters, elapsed time, and throughput. .github/workflows/benchmark.yml provides a manual equivalent. Normal pull-request CI generates 25,000 rows and does not run the expensive million-row benchmark.

Configuration

Variable	Default	Purpose
DATABASE_URL	sqlite:///./hospital_imports.db	PostgreSQL is used by Compose/production; SQLite is a developer fallback
REDIS_URL	redis://localhost:6379/0	Celery broker/result backend
IMPORT_STORAGE_PATH	./data/imports	Persistent source/chunk root
IMPORT_CHUNK_SIZE	10000	Rows per durable chunk
DATABASE_BATCH_SIZE	1000	Rows handled per database batch
WORKER_CONCURRENCY	8	Compose worker processes
MAX_RETRY_ATTEMPTS	5	Transient task retry ceiling
CHUNK_STALE_TIMEOUT_SECONDS	900	Abandoned claim threshold

Known bottlenecks

JSONL staging duplicates the uploaded data on local disk and requires shared durable storage when workers run on multiple hosts. Existing-key lookup plus ORM writes is portable and batched, but PostgreSQL COPY into a staging table followed by set-based ON CONFLICT would provide higher throughput. The status endpoint polls PostgreSQL; high-frequency clients should use caching or event delivery. A completed million-row run and real Redis worker interruption suite remain prerequisites for making a validated million-row claim.