FastSQS

The application object: register routes, attach routers and middleware, and handle Lambda events.

FastSQS

Bases: RecordProcessingMixin

Main FastSQS application class for handling AWS SQS messages.

FastAPI-style interface for routing and processing SQS messages with support for middleware, validation, and concurrency. Record/batch processing lives in RecordProcessingMixin (processing.py).

route

route(event_model: Type[SQSEvent], *, middlewares: Optional[List[Middleware]] = None) -> Callable[[Handler], Handler]

Register a route for a specific SQS event model.

Parameters:

Name	Type	Description	Default
event_model	Type[SQSEvent]	Pydantic model class for the event	required
middlewares	Optional[List[Middleware]]	Optional list of middlewares to apply	None

Returns:

Type	Description
Callable[[Handler], Handler]	Decorator function for the handler

default

default() -> Callable[[Handler], Handler]

Register a default handler for unmatched messages.

Returns:

Type	Description
Callable[[Handler], Handler]	Decorator function for the default handler

include_router

include_router(router: SQSRouter) -> None

Include an external router in the application.

Parameters:

Name	Type	Description	Default
router	SQSRouter	SQSRouter instance to include	required

add_middleware

add_middleware(middleware: Middleware) -> None

Add a middleware to the application.

Parameters:

Name	Type	Description	Default
middleware	Middleware	Middleware instance to add	required

handler

handler(event: dict | list, context: Any) -> dict

Main synchronous handler entry point for Lambda.

Parameters:

Name	Type	Description	Default
event	dict | list	SQS event — {"Records": [...]} or a bare list of records (the EventBridge Pipes target shape)	required
context	Any	Lambda context object	required

Returns:

Type	Description
dict	Dictionary with batch failure information

Raises:

Type	Description
RuntimeError	If called from within a running event loop

async_handler async

async_handler(event: dict | list, context: Any) -> dict

Asynchronous handler entry point for testing.

Parameters:

Name	Type	Description	Default
event	dict | list	SQS event — {"Records": [...]} or a bare list of records (the EventBridge Pipes target shape)	required
context	Any	Lambda context object	required

Returns:

Type	Description
dict	Dictionary with batch failure information