SQSRouter

Group routes across modules and nest them with subrouters.

SQSRouter

Router for handling SQS messages with multiple routing strategies.

Supports both pydantic model-based routing and key-value routing, with optional flexible message-type matching and nested subrouters.

route

route(value: Union[RouteValue, Iterable[RouteValue], Type[BaseModel], None] = None, *, model: Optional[type[BaseModel]] = None, middlewares: Optional[List[Middleware]] = None) -> Callable[[Handler], Handler]

Register a route handler.

Parameters:

Name	Type	Description	Default
value	Union[RouteValue, Iterable[RouteValue], Type[BaseModel], None]	Route value(s), a Pydantic model class, or None for the default	None
model	Optional[type[BaseModel]]	Optional Pydantic model for validation (key-value routes)	None
middlewares	Optional[List[Middleware]]	Optional list of middlewares	None

Returns:

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

Raises:

Type	Description
ValueError	If event model is invalid or a duplicate/colliding route exists

default

default(*, middlewares: Optional[List[Middleware]] = None) -> Callable[[Handler], Handler]

Register the default handler for messages that match no route.

Parameters:

Name	Type	Description	Default
middlewares	Optional[List[Middleware]]	Optional list of middlewares (currently advisory; the default handler runs under the router/app middleware chain).	None

Returns:

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

subrouter

subrouter(value: Union[RouteValue, Iterable[RouteValue]], router: Optional['SQSRouter'] = None) -> Union['SQSRouter', Callable[['SQSRouter'], 'SQSRouter']]

Register a subrouter for nested routing.

Parameters:

Name	Type	Description	Default
value	Union[RouteValue, Iterable[RouteValue]]	Route value(s) to associate with subrouter	required
router	Optional['SQSRouter']	Optional router instance	None

Returns:

Type	Description
Union['SQSRouter', Callable[['SQSRouter'], 'SQSRouter']]	Router instance or decorator function

add_middleware

add_middleware(mw: Middleware) -> None

Add middleware to this router.

Parameters:

Name	Type	Description	Default
mw	Middleware	Middleware instance to add	required

dispatch async

dispatch(payload: dict, record: dict, context: Any, ctx: 'Context', root_payload: Optional[dict] = None, parent_middlewares: Optional[List[Middleware]] = None) -> bool

Dispatch a message to the appropriate handler.

Parameters:

Name	Type	Description	Default
payload	dict	Message payload dictionary	required
record	dict	SQS record dictionary	required
context	Any	Lambda context object	required
ctx	'Context'	Per-record processing Context	required
root_payload	Optional[dict]	Original root payload	None
parent_middlewares	Optional[List[Middleware]]	Middlewares from parent routers	None

Returns:

Type	Description
bool	True if message was handled, False otherwise

Raises:

Type	Description
InvalidMessageError	If message validation fails