I have a FastAPI application where I would like to get the default logs written to the STDOUT with the following data in JSON format:
App logs should look like this:
{
"XYZ": {
"log": {
"level": "info",
"type": "app",
"timestamp": "2022-01-16T08:30:08.181Z",
"file": "api/predictor/predict.py",
"line": 34,
"threadId": 435454,
"message": "API Server started on port 8080 (development)"
}
}
}
Access logs should look like this:
{
"XYZ": {
"log": {
"level": "info",
"type": "access",
"timestamp": "2022-01-16T08:30:08.181Z",
"message": "GET /app/health 200 6ms"
},
"req": {
"url": "/app/health",
"headers": {
"host": "localhost:8080",
"user-agent": "curl/7.68.0",
"accept": "*/*"
},
"method": "GET",
"httpVersion": "1.1",
"originalUrl": "/app/health",
"query": {}
},
"res": {
"statusCode": 200,
"body": {
"statusCode": 200,
"status": "OK"
}
}
}
}
What I've tried:
I tried using the json-logging package for this. Using this example, I'm able to access the request logs in JSON and change the structure. But, I'm unable to find how to access and change the app logs.
Current output logs structure:
{"written_at": "2022-01-28T09:31:38.686Z", "written_ts": 1643362298686910000, "msg":
"Started server process [12919]", "type": "log", "logger": "uvicorn.error", "thread":
"MainThread", "level": "INFO", "module": "server", "line_no": 82, "correlation_id":
"-"}
{"written_at": "2022-01-28T09:31:38.739Z", "written_ts": 1643362298739838000, "msg":
"Started server process [12919]", "type": "log", "logger": "uvicorn.error", "thread":
"MainThread", "level": "INFO", "module": "server", "line_no": 82, "correlation_id":
"-"}
{"written_at": "2022-01-28T09:31:38.739Z", "written_ts": 1643362298739951000, "msg":
"Waiting for application startup.", "type": "log", "logger": "uvicorn.error",
"thread": "MainThread", "level": "INFO", "module": "on", "line_no": 45,
"correlation_id": "-"}
You could do that by creating a custom Formatter, using the built-in logger module in Python. You could use the extra parameter when logging messages to pass contextual information, such as url and headers.
Python's JSON module already implements pretty-printing JSON data, using the json.dumps() function and adjusting the indent level (one could instead use, likely faster, alternatives such as orjson).
Below is a working example using a custom formatter to log messages in a similar format to the one described in your question. For "app" logs, you could use, for instance, logger.info('sample log message'). Whereas, for "access" logs, you could use logger.info('sample log message', extra={'extra_info': get_extra_info(request, response)}). Passing the Request and Response instances to the get_extra_info() method would allow you extracting information about the request and response.
The approach below uses a FastAPI Middleware to log requests/responses, which allows you to handle the request before it is processed by some endpoint, as well as the response, before it is returned to the client. Additionally, the approach demonstrated below uses a BackgroundTask for logging the data (as described in this answer). A background task "will run only once the response has been sent" (as per Starlette documentation), meaning that the client won't have to wait for the logging to complete, before receiving the response. It is also worth mentioning that since logging using Python's built-in module is a synchronous operation, the write_log_data() background task is defined with normal def instead of async def, meaning that FastAPI will run it in a separate thread from anyio's external threadpool and then await it; otherwise, it would block the entire server, until is completed—for more details on this subject, as well as the various solutions available, please have a look at this answer. That is also the reason that the log output in the example below shows "thread_name": "AnyIO worker thread", when that background task was called after accessing /docs from a web browser.
For more LogRecord attributes, besides asctime, levelname, etc, have a look at the relevant documentation. The example below uses a RotatingFileHandler for writing the logs to a file on disk as well. If you don't need that, please feel free to remove it from the get_logger() method.
Finally, I would highly suggest having a look at this answer with regard to logging Request and Response data through a middleware, as well as this answer that demonstrates how to customize and use the uvicorn loggers too, in addition to logging your own custom messages. You might also want to consider initializing the logger within a lifespan handler at application startup, which would allow you to access the logger instance within APIRouter modules, outisde the main file of your application (which is usually the case when building Bigger Applications)—please have a look at this answer on how to achieve that.
app_logger.py
import logging, sys
def get_file_handler(formatter, filename="info.log", maxBytes=1024*1024, backupCount=3):
file_handler = logging.handlers.RotatingFileHandler(filename, maxBytes, backupCount)
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(formatter)
return file_handler
def get_stream_handler(formatter):
stream_handler = logging.StreamHandler(sys.stdout)
stream_handler.setLevel(logging.DEBUG)
stream_handler.setFormatter(formatter)
return stream_handler
def get_logger(name, formatter):
logger = logging.getLogger(name)
logger.setLevel(logging.DEBUG)
logger.addHandler(get_file_handler(formatter))
logger.addHandler(get_stream_handler(formatter))
return logger
app_logger_formatter.py
import logging, json
class CustomJSONFormatter(logging.Formatter):
def __init__(self, fmt):
logging.Formatter.__init__(self, fmt)
def format(self, record):
logging.Formatter.format(self, record)
return json.dumps(get_log(record), indent=2)
def get_log(record):
d = {
"time": record.asctime,
"process_name": record.processName,
"process_id": record.process,
"thread_name": record.threadName,
"thread_id": record.thread,
"level": record.levelname,
"logger_name": record.name,
"pathname": record.pathname,
#'filename': record.filename,
"line": record.lineno,
"message": record.message,
}
if hasattr(record, "extra_info"):
d["req"] = record.extra_info["req"]
d["res"] = record.extra_info["res"]
return d
app.py
from fastapi import FastAPI, Request, Response
from starlette.background import BackgroundTask
from app_logger_formatter import CustomJSONFormatter
from http import HTTPStatus
import app_logger
import uvicorn
app = FastAPI()
formatter = CustomJSONFormatter('%(asctime)s')
logger = app_logger.get_logger(__name__, formatter)
status_reasons = {x.value:x.name for x in list(HTTPStatus)}
def get_extra_info(request: Request, response: Response):
return {
"req": {
"url": request.url.path,
"headers": {
"host": request.headers["host"],
"user-agent": request.headers["user-agent"],
"accept": request.headers["accept"],
},
"method": request.method,
"http_version": request.scope["http_version"],
"original_url": request.url.path,
"query": {},
},
"res": {
"status_code": response.status_code,
"status": status_reasons.get(response.status_code),
}
}
def write_log_data(request, response):
logger.info(
request.method + " " + request.url.path,
extra={"extra_info": get_extra_info(request, response)}
)
@app.middleware("http")
async def log_request(request: Request, call_next):
response = await call_next(request)
response.background = BackgroundTask(write_log_data, request, response)
return response
@app.get("/")
async def foo(request: Request):
return "success"
if __name__ == '__main__':
logger.info("Server is listening...")
uvicorn.run(app, host='0.0.0.0', port=8000)
Output:
{
"time": "2024-10-27 12:15:00,115",
"process_name": "MainProcess",
"process_id": 1937,
"thread_name": "MainThread",
"thread_id": 1495,
"level": "INFO",
"logger_name": "__main__",
"pathname": "C:\\...",
"line": 56,
"message": "Server started listening on port: 8000"
}
{
"time": "2024-10-27 12:15:10,335",
"process_name": "MainProcess",
"process_id": 1937,
"thread_name": "AnyIO worker thread",
"thread_id": 1712,
"level": "INFO",
"logger_name": "__main__",
"pathname": "C:\\...",
"line": 37,
"message": "GET /docs",
"req": {
"url": "/docs",
"headers": {
"host": "127.0.0.1:8000",
"user-agent": "Mozilla...",
"accept": "text/html,application/xhtml+xml..."
},
"method": "GET",
"http_version": "1.1",
"original_url": "/docs",
"query": {}
},
"res": {
"status_code": 200,
"status": "OK"
}
}