Use when adding background tasks or event integration to a scaffolded FastAPI service after the broker and contracts are declared in backend-architecture. Adds Celery/RQ/arq or Kafka wiring, a transactional outbox, idempotent consumers, retry/DLQ, and Testcontainers tests. Not for the shell, auth, observability, or performance.
Use when adding background tasks or event integration to a scaffolded FastAPI service after the broker and contracts are declared in backend-architecture. Adds Celery/RQ/arq or Kafka wiring, a transactional outbox, idempotent consumers, retry/DLQ, and Testcontainers tests. Not for the shell, auth, observability, or performance.
FastAPI Async and Task Integration
When to use
Invoke when a scaffolded FastAPI service must enqueue background work, publish domain events, or consume from a broker (Celery/RQ/arq on Redis, or Kafka), and the integration needs correct delivery semantics, idempotency, retry/DLQ, and an outbox so a database commit and an event publish cannot diverge.
Do not use for: the service shell, settings, or error tiers (use fastapi-service-scaffold), auth or OWASP review (use fastapi-auth-and-security-review), OpenTelemetry/metrics/SLO wiring (use fastapi-observability-readiness), or pool-sizing/circuit-breaker/load-test gating (use fastapi-performance-and-resilience).
Inputs
Required:
A service with the fastapi-service-scaffold baseline (DI providers, validated settings, logging, ASGI-lifespan shutdown present).
backend-architecture.md declaring the broker, the event/message contracts, and the delivery-semantics and retry strategy — or explicit confirmation a needed decision is intentionally deferred.
Optional:
Approved architecture/reliability for redelivery/DLQ expectations and the consumer SLO.
The transactional data store (for the outbox table) if backend-architecture.md is silent.
Ordering and partitioning requirements (Kafka key, queue priority) per contract.
Operating rules
Never invent the broker, contract, or delivery semantics. Broker choice, message/event schemas, ordering, and at-least-once vs effectively-once belong to backend-architecture.md. If silent on a decision this skill needs, pause and raise an ADR candidate rather than guessing.
Extend the scaffold; do not duplicate it. Producers/consumers register in the scaffold DI providers, read connection settings from the validated settings seam, log through the scaffold structlog logger, and close via the scaffold ASGI-lifespan shutdown. Do not re-create any of these.
Delivery is at-least-once unless the broker and contract guarantee otherwise: therefore every consumer/task is idempotent. Idempotency is explicit — a dedupe key (message id or business key) checked against a store, not "the task is probably safe to re-run".
A database write that must produce an event uses the transactional outbox: the domain change and the outbox row commit in one transaction; a relay publishes from the outbox. Never publish inside the request path before the transaction commits (dual-write hazard).
Failure handling is explicit and bounded: a retry policy with backoff and a max attempt count, then a dead-letter destination. A message/task is never retried forever and never silently dropped.
Workers respect shutdown: on SIGTERM the worker stops fetching, finishes in-flight tasks within a bounded timeout, and does not ack work it did not complete. Poison messages go to the DLQ, not an infinite redelivery loop.
Producers do not block the request path on broker latency beyond a bounded timeout; a broker outage degrades to the outbox (for transactional events) or a clear, handled error — never an unbounded hang or a synchronous broker call on the async path.
An integration without a real-broker test is not done. Provide Testcontainers-backed tests proving: successful round trip, duplicate delivery handled idempotently, retry then DLQ on poison, and clean shutdown mid-consume.
A change that does not pass mypy, ruff, the integration tests, and the boot smoke check is not done. Fix and re-run.
Output contract
The async/task integration MUST conform to:
api-standards — message/event payloads match the declared contract and are schema-validated (Pydantic) on produce and consume; versioned, explicit envelope.
observability-standards — produce/consume go through the scaffold logger with correlation; retry, DLQ, and lag are observable.
Upstream contract: backend-architecture.md is the source of truth for broker, contracts, ordering, and delivery semantics; architecture/reliability is the source of truth for redelivery/DLQ expectations and the consumer SLO. If either is silent on a decision this skill needs, pause and raise an ADR candidate rather than guessing.
Progressive references
Read references/fastapi-async-playbook.md when implementing any owned area or checking the anti-pattern list.
Read references/fastapi-async-quality-rubric.md before declaring the work complete.
Use assets/fastapi-async-and-task-integration.template.md as the producer, consumer, outbox, and test reference.
Process
Gather context: load backend-architecture.md (broker, contracts, ordering, delivery semantics) and architecture/reliability (redelivery/DLQ, consumer SLO). Confirm the scaffold baseline and the transactional store. If a needed decision is missing, raise an ADR candidate before proceeding.
Extend settings: add broker connection settings, queue/topic names, worker concurrency, retry attempts/backoff, and DLQ target to the scaffold Settings model and .env.example (placeholders only).
Define the message envelope and schemas: a versioned Pydantic envelope (id, type, occurred_at, schema_version, payload) with a model per message type, validated on both produce and consume; reject unknown types.
Implement the producer: a typed publish API registered in the scaffold DI providers, with a bounded send timeout. For events tied to a database write, write to an outbox table in the same transaction instead of publishing inline.
Implement the transactional outbox relay: a poller (or CDC hook) that reads unsent outbox rows, publishes them, marks them sent, and is itself idempotent and at-least-once safe.
Implement the consumer/task: registered in the DI providers, idempotent via an explicit dedupe-key store, with a bounded retry+backoff policy and a max-attempt threshold that routes to the DLQ. Honor the scaffold ASGI-lifespan shutdown (stop fetching, drain in-flight, no ack of incomplete work).
Make it observable: produce, consume, retry, DLQ, and consumer lag emit through the scaffold logger with correlation; expose the metrics seam hooks for fastapi-observability-readiness to instrument (do not wire the vendor here).
Write integration tests with Testcontainers: a real broker container; assert successful round trip, idempotent handling of a duplicate, retry-then-DLQ on a poison message, and clean shutdown mid-consume without losing or double-acking work.
Build verification (mandatory): run mypy, ruff check, the integration test command, and the boot smoke check. Fix and re-run on failure. Validate against the Output contract standards; document any unresolved gap in the service README.
Outputs
Required:
Typed producer registered in the scaffold DI providers with a bounded send timeout.
Transactional outbox table + relay for events tied to a database write (no inline dual write).
Idempotent consumer/task with an explicit dedupe-key store, bounded retry+backoff, and a DLQ route.
Versioned Pydantic message envelope with per-type models validated on produce and consume.
Shutdown-aware worker wired to the scaffold ASGI-lifespan shutdown.
