TraceId is a trace ID generator based on Python contextvars. It automatically passes the trace ID in asynchronous tasks, enabling cross-coroutine tracing.
pip install traceid
poetry add traceid
from traceid import TraceId
TraceId.gen()
from traceid import TraceId
TraceId.set('your trace id')
from traceid import TraceId
TraceId.get()
from traceid import TraceId
TraceId.clear()
from traceid import TraceId
TraceId.is_set()
import typing
from fastapi import FastAPI, Request, Response
from traceid import TraceId
from aiorow import Connection, DSN
app = FastAPI()
@app.middleware("http")
async def add_trace_id(
request: Request, call_next: typing.Callable[[Request], typing.Awaitable[Response]]
) -> Response:
trace_id = request.headers.get("X-Request-ID", None)
if trace_id is None:
TraceId.gen()
else:
TraceId.set(trace_id)
response = await call_next(request)
response.headers["X-Request-ID"] = str(TraceId.get())
return response
@app.get("/")
def read_root():
return {"Hello": "World"}
import logging
import logging.config
import os
from traceid import JSONFormatter
LOG_SETTINGS = {
"version": 1,
"disable_existing_loggers": False,
"handlers": {
"console": {
"class": "logging.StreamHandler",
"level": logging.INFO,
"formatter": "json",
},
},
"formatters": {
"json": {
"()": JSONFormatter,
},
},
"loggers": {
"": {"level": logging.INFO, "handlers": ["console"], "propagate": True},
},
}
logging.config.dictConfig(LOG_SETTINGS)