You could do that by creating a custom Formatter using the built-in logger module. You can 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 in the dump() function, using the indent parameter that allows you to define the indent level. Below is a working example using a custom formatter to log messages in the format you described in your question. For "App" logs use, for instance, logger.info('sample log message'), whereas, for "Access" logs, use logger.info('sample log message', extra={'extra_info': get_extra_info(request)}). By passing the request instance to the get_extra_info() method, you can extract information such as the one you mentioned above. For more LogRecord attributes, have a look here. The below example uses a FileHandler for logging the messages to a log file on disk as well. If you don't need that, you can comment it out in the get_logger() method.
The approach below uses a FastAPI Middleware to log requests/responses, which allows you to handle the request before it is processed by some specific 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 client won't have to be waiting for the logging to complete before receiving the response. See this related answer as well.
app_logger.py
import logging, sys
def get_file_handler(formatter, log_filename):
file_handler = logging.FileHandler(log_filename)
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, log_filename = "logfile.log"):
logger = logging.getLogger(name)
logger.setLevel(logging.DEBUG)
logger.addHandler(get_file_handler(formatter, log_filename))
logger.addHandler(get_stream_handler(formatter))
return logger
app_logger_formatter.py
import json, logging
def get_app_log(record):
json_obj = {'XYZ': {'log': {
'level': record.levelname,
'type': 'app',
'timestamp': record.asctime,
#'filename': record.filename,
'pathname': record.pathname,
'line': record.lineno,
'threadId': record.thread,
'message': record.message
}}}
return json_obj
def get_access_log(record):
json_obj = {'XYZ': {'log': {
'level': record.levelname,
'type': 'access',
'timestamp': record.asctime,
'message': record.message},
'req': record.extra_info['req'],
'res': record.extra_info['res']}}
return json_obj
class CustomFormatter(logging.Formatter):
def __init__(self, formatter):
logging.Formatter.__init__(self, formatter)
def format(self, record):
logging.Formatter.format(self, record)
if not hasattr(record, 'extra_info'):
return json.dumps(get_app_log(record), indent=2)
else:
return json.dumps(get_access_log(record), indent=2)
app.py
import app_logger
from app_logger_formatter import CustomFormatter
from fastapi import FastAPI, Request, Response
from http import HTTPStatus
from starlette.background import BackgroundTask
import uvicorn
app = FastAPI()
formatter = CustomFormatter('%(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,
'httpVersion': request.scope['http_version'],
'originalUrl': request.url.path,
'query': {}
},
'res': {'statusCode': response.status_code, 'body': {'statusCode': 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("/foo")
def foo(request: Request):
return "success"
if __name__ == '__main__':
logger.info("Server started listening on port: 8000")
uvicorn.run(app, host='0.0.0.0', port=8000)
Output:
{
"XYZ": {
"log": {
"level": "INFO",
"type": "app",
"timestamp": "2022-01-28 10:46:01,904",
"pathname": ".../app.py",
"line": 33,
"threadId": 408,
"message": "Server started listening on port: 8000"
}
}
}
{
"XYZ": {
"log": {
"level": "INFO",
"type": "access",
"timestamp": "2022-01-28 10:46:03,587",
"message": "GET /foo"
},
"req": {
"url": "/foo",
"headers": {
"host": "127.0.0.1:8000",
"user-agent": "Mozilla/5.0 ...",
"accept": "text/html,..."
},
"method": "GET",
"httpVersion": "1.1",
"originalUrl": "/foo",
"query": {}
},
"res": {
"statusCode": 200,
"body": {
"statusCode": 200,
"status": "OK"
}
}
}
}
