Life Cycle Hooks#
Life cycle hooks allow the execution of a callable at a certain point during the request-response cycle. The hooks available are:
Name |
Runs |
---|---|
Before the router handler function |
|
After the route handler function |
|
After the response has been sent |
Before Request#
The before_request
hook runs immediately before calling the route handler function. It
can be any callable accepting a Request
as its first parameter
and returns either None
or a value that can be used in a response.
If a value is returned, the router handler for this request will be bypassed.
from typing import Dict, Optional
from litestar import Litestar, Request, get
async def before_request_handler(request: Request) -> Optional[Dict[str, str]]:
name = request.query_params["name"]
if name == "Ben":
return {"message": "These are not the bytes you are looking for"}
request.state["message"] = f"Use the handler, {name}"
return None
@get("/")
async def handler(request: Request, name: str) -> Dict[str, str]:
message: str = request.state["message"]
return {"message": message}
app = Litestar(route_handlers=[handler], before_request=before_request_handler)
from typing import Optional
from litestar import Litestar, Request, get
async def before_request_handler(request: Request) -> Optional[dict[str, str]]:
name = request.query_params["name"]
if name == "Ben":
return {"message": "These are not the bytes you are looking for"}
request.state["message"] = f"Use the handler, {name}"
return None
@get("/")
async def handler(request: Request, name: str) -> dict[str, str]:
message: str = request.state["message"]
return {"message": message}
app = Litestar(route_handlers=[handler], before_request=before_request_handler)
from litestar import Litestar, Request, get
async def before_request_handler(request: Request) -> dict[str, str] | None:
name = request.query_params["name"]
if name == "Ben":
return {"message": "These are not the bytes you are looking for"}
request.state["message"] = f"Use the handler, {name}"
return None
@get("/")
async def handler(request: Request, name: str) -> dict[str, str]:
message: str = request.state["message"]
return {"message": message}
app = Litestar(route_handlers=[handler], before_request=before_request_handler)
Run it
> curl http://127.0.0.1:8000/?name=Luke
{"message":"Use the handler, Luke"}
> curl http://127.0.0.1:8000/?name=Ben
{"message":"These are not the bytes you are looking for"}
After Request#
The after_request
hook runs after the route handler returned and the response object
has been resolved. It can be any callable which takes a Response
instance as its first parameter, and returns a Response
instance. The Response
instance returned does not necessarily have to be the one that was received.
from typing import Dict
from litestar import Litestar, MediaType, Response, get
async def after_request(response: Response) -> Response:
if response.media_type == MediaType.TEXT:
return Response({"message": response.content})
return response
@get("/hello")
async def hello() -> str:
return "Hello, world"
@get("/goodbye")
async def goodbye() -> Dict[str, str]:
return {"message": "Goodbye"}
app = Litestar(route_handlers=[hello, goodbye], after_request=after_request)
from litestar import Litestar, MediaType, Response, get
async def after_request(response: Response) -> Response:
if response.media_type == MediaType.TEXT:
return Response({"message": response.content})
return response
@get("/hello")
async def hello() -> str:
return "Hello, world"
@get("/goodbye")
async def goodbye() -> dict[str, str]:
return {"message": "Goodbye"}
app = Litestar(route_handlers=[hello, goodbye], after_request=after_request)
Run it
> curl http://127.0.0.1:8000/hello
{"message":"Hello, world"}
> curl http://127.0.0.1:8000/goodbye
{"message":"Goodbye"}
After Response#
The after_response
hook runs after the response has been returned by the server.
It can be any callable accepting a Request
as its first parameter
and does not return any value.
This hook is meant for data post-processing, transmission of data to third party services, gathering of metrics, etc.
from collections import defaultdict
from typing import Dict
from litestar import Litestar, Request, get
COUNTER: Dict[str, int] = defaultdict(int)
async def after_response(request: Request) -> None:
COUNTER[request.url.path] += 1
@get("/hello")
async def hello() -> Dict[str, int]:
return COUNTER
app = Litestar(route_handlers=[hello], after_response=after_response)
from collections import defaultdict
from litestar import Litestar, Request, get
COUNTER: dict[str, int] = defaultdict(int)
async def after_response(request: Request) -> None:
COUNTER[request.url.path] += 1
@get("/hello")
async def hello() -> dict[str, int]:
return COUNTER
app = Litestar(route_handlers=[hello], after_response=after_response)
Run it
> curl http://127.0.0.1:8000/hello
{}
> curl http://127.0.0.1:8000/hello
{"/hello":1}
Note
Since the request has already been returned by the time the after_response
is called,
the updated state of COUNTER
is not reflected in the response.
Layered hooks#
Layered architecture
Life cycle hooks are part of Litestar’s layered architecture, which means you can set them on every layer of the application. If you set hooks on multiple layers, the layer closest to the route handler will take precedence.
You can read more about this here: Layered architecture
from litestar import Litestar, Response, get
def after_request_app(response: Response) -> Response:
return Response(content=b"app after request")
def after_request_handler(response: Response) -> Response:
return Response(content=b"handler after request")
@get("/")
async def handler() -> str:
return "hello, world"
@get("/override", after_request=after_request_handler)
async def handler_with_override() -> str:
return "hello, world"
app = Litestar(
route_handlers=[handler, handler_with_override],
after_request=after_request_app,
)
Run it
> curl http://127.0.0.1:8000/
app after request
> curl http://127.0.0.1:8000/override
handler after request