Middleware

The middleware base class and the built-in middleware.

Middleware

Base class for FastSQS middleware.

Middleware can hook into message processing before and after handler execution. Subclasses override :meth:before and/or :meth:after.

before async

before(payload: dict, record: dict, context: Any, ctx: 'Context') -> None

Hook called before handler execution.

Parameters:

Name	Type	Description	Default
payload	dict	Message payload	required
record	dict	SQS record	required
context	Any	Lambda context	required
ctx	'Context'	Per-record processing Context	required

Raising from before aborts processing for this record (the handler does not run); already-entered middlewares are still unwound via after.

after async

after(payload: dict, record: dict, context: Any, ctx: 'Context', error: Optional[Exception]) -> None

Hook called after handler execution.

Parameters:

Name	Type	Description	Default
payload	dict	Message payload	required
record	dict	SQS record	required
context	Any	Lambda context	required
ctx	'Context'	Per-record processing Context	required
error	Optional[Exception]	Exception if the handler (or a before hook) failed, else None	required

TimingMiddleware

Bases: Middleware

Middleware that measures message processing duration.

Records a start time before processing and stores the duration (ms) in ctx.state after, so downstream middleware/handlers can read it.

before async

before(payload, record, context, ctx)

Record processing start time in ctx.state.

after async

after(payload, record, context, ctx, error)

Compute and store processing duration in ctx.state.

LoggingMiddleware

Bases: Middleware

Middleware that provides structured logging for message processing.

Logs detailed information about message processing including payloads, timing, errors, and processing context with field masking support. Defaults to JSON-line logging on stdout (CloudWatch-friendly), no external dependencies.

log

log(level: str, message: str, **data: Any) -> None

Log a message with structured data.

Parameters:

Name	Type	Description	Default
level	str	Log level	required
message	str	Log message	required
**data	Any	Additional structured data	{}

before async

before(payload, record, context, ctx)

Log message processing start with context information.

after async

after(payload, record, context, ctx, error)

Log message processing completion with results and errors.

Idempotency

IdempotencyMiddleware

Bases: Middleware

Dedup middleware over an :class:IdempotencyStore.

The key is resolved from the payload via key_path (dot-paths traverse nested dicts, e.g. "metadata.eventId"; default "id", matching CloudEvents) and exposed to handlers as ctx.state.idempotency_key.

On success (or a handler-raised :class:~fastsqs.SkipMessage, which acks) the key is marked completed for completed_ttl_seconds; on failure the claim is released so the SQS redelivery retries.

IdempotencyStore

Bases: Protocol

Structural port for idempotency state.

Any object with these three async methods satisfies it — no fastsqs base class or inheritance required (DynamoDB conditional puts, Redis SET NX, Postgres upserts...). try_acquire MUST be atomic in the backing store: two concurrent callers with the same key must never both get ACQUIRED.

try_acquire async

try_acquire(key: str, ttl_seconds: int) -> AcquireResult

Atomically claim key for ttl_seconds (an IN_PROGRESS lease).

Returns ACQUIRED on a fresh claim (including one whose previous lease/window expired), else the live entry's state.

mark_complete async

mark_complete(key: str, ttl_seconds: int) -> None

Mark key processed; duplicates skip for ttl_seconds (the dedup window).

forget async

forget(key: str) -> None

Release key after a failed attempt so a redelivery can retry.

InMemoryIdempotencyStore

Reference :class:IdempotencyStore for tests and single-process dev.

State lives in the process — it does NOT survive restarts and is not shared across Lambda sandboxes; use a DynamoDB/Redis-backed store in production. Methods contain no await between check and write, so they are atomic under a single event loop.

clock is injectable (monotonic seconds) so tests can control expiry.

AcquireResult

Bases: Enum

Outcome of :meth:IdempotencyStore.try_acquire.

Tracing

TracingMiddleware

Bases: Middleware

Expose the record's W3C trace context as ctx.state.trace.

Sources, in precedence order (message attributes are the transport-level channel, so they win): SQS message attributes, then top-level payload keys (the CloudEvents extension-attribute convention). When absent or invalid, ctx.state is left untouched (ctx.state.get("trace") -> None).

TraceContext dataclass

Parsed W3C traceparent (plus the opaque tracestate, verbatim).