Adding Context to each FastAPI request using request state
Take anything from loggers to data to functions straight from your request in your endpoints
In this article we’ll create a Context object that you can inject into each request. We’ll use it to add a unique ID and pre-configured logger to each request so so that we can add it to our log messages and differentiate them from each other, like in the image below:
In the image above you can see path that the request with id 20241222T123410938110_qvrJEXMv took through our API.
After some prep we'll see that it's pretty easy to implement the Context object. Additionally, the Context object can (and will in future articles!) be expanded to suit other needs like timer-functions, request-specific data etc. Let’s code!
Setup: API with a simple route
First we'll need an API, which is pretty easy to set up:
import random
import time
from fastapi import FastAPI, Depends
from request_context import Ctx, get_request_ctx
from loguru import logger
app = FastAPI(title="My API")
@app.get('/process')
def test_route(amount:int, ctx: Ctx = Depends(get_request_ctx)) -> Optional[float]:
logger.debug(f"Performing task 1 with {amount=}")
time.sleep(random.randrange(10, 50) / 100) # sleep 0.1 - 0.5 seconds
logger.debug("Performing task 2")
time.sleep(random.randrange(10, 50) / 100) # sleep 0.1 - 0.5 seconds
logger.debug("Performing task 3")
try:
result = 100 / amount
except Exception as e:
logger.error(f"error: {e}")
return None
logger.info("Task completed!")The main thing to take into account here is that our process route accepts one parameter: amount. Then it performs three tasks. For each we'll produce a log message and then sleep a while to simulate some work. In the last task we divide the number 100 by the amount and return the result.
Mike Huls
Setup: Simulating heavy traffic
In order to simulate heavy traffic on our endpoint, we'll create a piece of code that sends a number of requests simultaneously, each with a different amount (in the case of the code below 0, 1, 2, 3 and 4).
import asyncio
import httpx
async def send_request(amount:int):
async with httpx.AsyncClient() as client:
response = await client.get(f"http://localhost:8000/process?amount={amount}")
print(response.status_code, response.headers)
async def main(_n_requests:int):
tasks = [send_request(i) for i in range(_n_requests)]
await asyncio.gather(*tasks)
asyncio.run(main(5))When we execute this code, our API outputs the following:
This is pretty nice, we have a lot of information but we have no idea which specific request causes the division-error. It would be nice to see which log belongs to which request so that we can connect logs together!
Adding state to our request using Middleware
In order to solve our challenge we'll create a piece of middleware for our API. This intercepts all traffic on API and allows you to execute code before and after the request is passed on to the endpoint. It looks like this:
@app.middleware("http")
async def add_request_context(request: Request, call_next:Callable):
request.state.context = Ctx()
return await call_next(request)These 4 lines of code will listen to any http-traffic on our router and, before it passes the request further, puts a Ctx object on the state of the request under the context key. Think of the request.state as a dictionary where you can store things.
1. Add information to the state of our request
The Ctx will hold all the context that we want to save for our request:
import loguru
@dataclass
class Ctx:
correlation_id: str
logger: Logger
def __init__(self):
self.correlation_id = uuid.uuid4().hex
_logger = loguru.logger
self.logger = _logger.bind(correlation_id=self.correlation_id)
As you see it will create and store a correlation_id; this is the id that is unique for each request since each request will have their own instance of Ctx.
Next Ctx will create a loguru logger that we'll bind the correlation_id to. We don't particularly need loguru but I like how easily configurable it is. Also binding a correlation ID is pretty convenient.
If you don't like to use loguru you can easily modify Ctx to work with python's default loging library e.g.
Mike Huls
2. Configuring our logger
Now that our logger is "aware" of the correlation-id (using logger.bind), we can modify the format of our logger so that it prints the logs with the correlation-id to our console. This is achieved with the following:
logger.remove()
logger.add(
sink=sys.stderr,
level="TRACE" ,
format="<green>{extra[correlation_id]} {level.icon}{time:HH:MM:SS} 📁{module:<10} 🤖{function:<10} 🔢{line:<3}</green> <level>{message}</level>",
)Notice that in our format we use {extra[correlation_id]}. This is because _logger.bind(correlation_id=self.correlation_id) that we've handled in the previous part will store the value in the extra of the log under the provided correlation_id key.
3. Retrieving the context from the request
We'd like to use the logger in our route so we'll need to retrieve it from the context. Luckily this is pretty easy because FastAPI injects each request in each route:
@app.get('/process2')
def test_route(request:Request, amount: int) -> Optional[float]:
ctx = request.state.context
ctx.logger.debug(f"Performing task 1 with {amount=}")
time.sleep(random.randrange(10, 50) / 100) # sleep 0.1 - 0.5 seconds
... [rest of the route]Here you see that we can just extract the context from the request. Now that we have access to our Ctx instance, we can use its logger. In the previous part we've configured it to log its message along with the correlation_id that was bound to it. Let's see the results:
In this overview I've highlighted the log that has the error. We can easily see what path it took and find out that its amount was 0, leading to the division by zero error.
Mike Huls
Conclusion
This article demonstrated how to create a Ctx object type to add to the request state. This object encapsulates all functionalities that allow you to use the object in your functions. In the examples we’ve used the context object to improve our logs with request-specific ID’s but the possibilities are endless. I hope to have demonstrated a clear way to manage request context to improve your API.
I hope this article was as clear as I intended it to be but if this is not the case please let me know what I can do to clarify further. In the meantime, check out my other articles on all kinds of programming-related topics.
Happy coding!
— Mike
P.s: like what I'm doing? Follow me!
