Middlewares
Description
Middleware is an intermediate software layer that allows performing specific actions before and after processing a request by a route.
from rapidy import Rapidy
from rapidy.http import middleware, get, Request, StreamResponse
from rapidy.typedefs import CallNext
@middleware
async def hello_rapidy_middleware(request: Request, call_next: CallNext) -> StreamResponse:
print('before')
handler_response = await call_next(request)
print('after')
return handler_response
@get('/')
async def handler() -> dict[str, str]:
return {'hello': 'rapidy'}
rapidy = Rapidy(
http_route_handlers=[handler],
middlewares=[hello_rapidy_middleware],
)
Attribute Order
In a web handler, the first argument is always Request, and the second is CallNext (the request handler or middleware in the chain).
(They have the same signature — rapidy.typedefs.CallNext.)
Handler Response Type
The response from a web handler or the next middleware in the chain will always be StreamResponse. Keep this in mind when designing your middleware.
Creation Methods
There are two ways to create middleware.
Without Parameters
This approach is suitable if you do not need to manage the request within the middleware.
With Parameters
Use this approach if you want more flexibility in managing the response to a request.
@middleware(
response_validate=...,
response_type = ...,
response_content_type = ...,
response_charset = ...,
response_zlib_executor = ...,
response_zlib_executor_size = ...,
response_include_fields = ...,
response_exclude_fields = ...,
response_by_alias = ...,
response_exclude_unset = ...,
response_exclude_defaults = ...,
response_exclude_none = ...,
response_custom_encoder = ...,
response_json_encoder = ...,
)
async def hello_middleware(request: Request, call_next: CallNext) -> StreamResponse:
Attributes
Rapidy-middleware supports all data validation mechanisms available in web handlers, as well as response handling.
Validation
Like web handlers, middleware can access request objects via attributes.
Before proceeding, we recommend reviewing the section Request — Managing HTTP Requests since middleware follows the same logic for handling request parameters.
Processing a Bearer token.
from rapidy.http import middleware, StreamResponse, Header, Request
from rapidy.typedefs import CallNext
TOKEN_REGEXP = '^[Bb]earer (?P<token>[A-Za-z0-9-_=.]*)'
@middleware
async def get_bearer_middleware(
request: Request,
call_next: CallNext,
bearer_token: str = Header(alias='Authorization', pattern=TOKEN_REGEXP),
) -> StreamResponse:
# process token here ...
return await call_next(request)
If you extract the body both in middleware and in the handler, you will not encounter an error about data being read multiple times.
Extracted data is cached in memory and reused during validation.
Response Management
Like web handlers, middleware can manage responses using its own attributes.
Before proceeding, we recommend reviewing the section Response — Managing HTTP Responses since middleware follows the same response handling logic.
Response management is only possible if the middleware returns an unprocessed data type (anything other than Response or StreamResponse).
Middleware manages the response using attributes.
from rapidy import Rapidy
from rapidy.http import middleware, StreamResponse, get, Request
from rapidy.enums import ContentType
from rapidy.typedefs import CallNext
@middleware(response_content_type=ContentType.text_html)
async def hello_rapidy_middleware(request: Request, call_next: CallNext) -> StreamResponse | str:
try:
return await call_next(request)
except Exception:
return 'server error' # Content-Type='text/html'
@get('/')
async def handler() -> dict[str, str]:
raise Exception
rapidy = Rapidy(middlewares=[hello_rapidy_middleware], http_route_handlers=[handler])
Middleware cannot manage the response using attributes.
from rapidy import Rapidy
from rapidy.http import middleware, StreamResponse, get, Request, Response
from rapidy.enums import ContentType
from rapidy.typedefs import CallNext
@middleware(response_content_type=ContentType.text_html)
async def hello_rapidy_middleware(request: Request, call_next: CallNext) -> StreamResponse:
try:
return await call_next(request)
except Exception:
return Response(status=500) # Content-Type='application/octet-stream'
@get('/')
async def handler() -> dict[str, str]:
raise Exception
rapidy = Rapidy(middlewares=[hello_rapidy_middleware], http_route_handlers=[handler])
Accessing Response.
@middleware
async def hello_rapidy_middleware(
request: Request,
call_next: CallNext,
response: Response,
) -> StreamResponse:
Response is created only for the current middleware.
Middleware Returns a Different Data Type
If middleware returns any data type other than StreamResponse, specify this type in Union so that Rapidy can use it for response validation.