FastSQS turns an SQS-triggered Lambda into a typed, declarative app. You write handlers for pydantic event models; FastSQS parses each record, routes it, validates it, runs your middleware, and returns the batchItemFailures SQS expects, so failed messages are redelivered and dead-lettered by the queue's own redrive policy, not by bespoke in-app code.

Features

• FastAPI-style routing: @app.route(OrderCreated) dispatches by a payload discriminator (default key "type").
• Pydantic validation: handlers receive a validated SQSEvent model; bad messages become clean batch failures.
• Standard envelopes: opt-in CloudEvent[T] base (CloudEvents 1.0), dot-path discriminators for metadata+data envelopes, and __message_type__ for namespaced/versioned route keys (com.acme.payment.approved.v1).
• Dependency injection: declare Depends(...) params (powered by fast-depends); no @inject needed.
• Typed Context: ctx.message_id, ctx.queue_type, and more as typed attributes; arbitrary scratch in ctx.state.
• Middleware: before/after hooks with balanced unwind (resources acquired in before are always released).
• Idempotency: IdempotencyMiddleware dedups at-least-once delivery behind a structural IdempotencyStore protocol (in-memory store included); raise SkipMessage anywhere to ack a record without processing.
• Tracing: TracingMiddleware reads W3C traceparent from message attributes into a typed ctx.state.trace.
• Partial batch failure: native ReportBatchItemFailures for standard and FIFO queues.
• FIFO-aware: queue type is inferred from the event-source ARN; per-group ordering with a configurable failure mode.
• EventBridge Pipes ready: app.handler accepts both the Lambda {"Records": [...]} envelope and a bare list of records.
• Shape detection: is_sqs_event(event) lets one Lambda multiplex SQS and non-SQS (e.g. API Gateway) events.
• In-process test client: drive your app with synthetic events, no AWS required.
• Typed: ships py.typed; full editor and mypy support.

Install

pip install fastsqs

Requires Python 3.10+. Depends on pydantic>=2 and fast-depends>=3,<4.

Quick start

from fastsqs import FastSQS, SQSEvent

app = FastSQS()  # queue type auto-detected from the event-source ARN


class OrderCreated(SQSEvent):
    order_id: str
    amount: int


@app.route(OrderCreated)
async def handle_order(msg: OrderCreated):
    print("processing", msg.order_id, msg.amount)
    # raising marks this record as failed -> SQS redelivers it


# Lambda entry point (set as the function handler):
def handler(event, context):
    return app.handler(event, context)

A message routes by its discriminator value ("type" by default), matched to the event model's name in snake_case, so {"type": "order_created", ...} routes to OrderCreated. Field names accept both snake_case and their camelCase aliases via Pydantic alias generation.

Why fastsqs

Reach for fastsqs when a queue carries many message types and you would otherwise branch by hand in one large handler. It gives you correct ReportBatchItemFailures together with the FastAPI model on top: route by type, validate with pydantic, inject dependencies, and read a typed Context. For a single trivial handler with no validation, a plain boto3 loop is still fine.

Documentation

Full documentation, including guides, concepts, and the API reference: fastsqs.github.io

• Quickstart
• Guide — routing, dependency injection, middleware, FIFO, testing
• Concepts — how routing, the batch lifecycle, and partial batch failure work
• API Reference

Contributing

Issues and pull requests are welcome. See CONTRIBUTING.md for dev setup, and the Code of Conduct. Open an issue to discuss anything non-trivial first.

License

MIT. See LICENSE.