Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""Client middleware support."""
3from collections.abc import Awaitable, Callable, Sequence
5from .client_reqrep import ClientRequest, ClientResponse
7__all__ = ("ClientMiddlewareType", "ClientHandlerType", "build_client_middlewares")
9# Type alias for client request handlers - functions that process requests and return responses
10ClientHandlerType = Callable[[ClientRequest], Awaitable[ClientResponse]]
12# Type for client middleware - similar to server but uses ClientRequest/ClientResponse
13ClientMiddlewareType = Callable[
14 [ClientRequest, ClientHandlerType], Awaitable[ClientResponse]
15]
18def build_client_middlewares(
19 handler: ClientHandlerType,
20 middlewares: Sequence[ClientMiddlewareType],
21) -> ClientHandlerType:
22 """
23 Apply middlewares to request handler.
25 The middlewares are applied in reverse order, so the first middleware
26 in the list wraps all subsequent middlewares and the handler.
28 This implementation avoids using partial/update_wrapper to minimize overhead
29 and doesn't cache to avoid holding references to stateful middleware.
30 """
31 # Optimize for single middleware case
32 if len(middlewares) == 1:
33 middleware = middlewares[0]
35 async def single_middleware_handler(req: ClientRequest) -> ClientResponse:
36 return await middleware(req, handler)
38 return single_middleware_handler
40 # Build the chain for multiple middlewares
41 current_handler = handler
43 for middleware in reversed(middlewares):
44 # Create a new closure that captures the current state
45 def make_wrapper(
46 mw: ClientMiddlewareType, next_h: ClientHandlerType
47 ) -> ClientHandlerType:
48 async def wrapped(req: ClientRequest) -> ClientResponse:
49 return await mw(req, next_h)
51 return wrapped
53 current_handler = make_wrapper(middleware, current_handler)
55 return current_handler