How do I get my FastAPI application's console log in JSON format with a different structure and different fields?

Question

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": "-"}

Answer 1

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 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 in a log file as well. If you don't need that, you can comment it out in the get_logger() method.

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
import uvicorn

app = FastAPI()
formatter = CustomFormatter('%(asctime)s')
logger = app_logger.get_logger(__name__, formatter)


def get_extra_info(request):
    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': 200, 'body': {'statusCode': 200,
                   'status': 'OK'}}}


@app.get("/foo")
def foo(request: Request):
    logger.info('GET ' + request.url.path, extra={'extra_info': get_extra_info(request)})
    return "success"


if __name__ == '__main__':
    logger.info("Server started listening on port: 8000")
    uvicorn.run(app, host='127.0.0.1', port=8000)

UPDATE 1

A better approach to log requests/responses is to add a FastAPI Middleware, which takes each request that comes to your application, as well as every response before returning it. Example is given below (app_logger.py and app_logger_formatter.py remain the same as above).

app.py

import app_logger
from app_logger_formatter import CustomFormatter
from fastapi import FastAPI, Request, Response
import uvicorn
from http import HTTPStatus

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)}}}


@app.middleware("http")
async def log_request(request: Request, call_next):
    response = await call_next(request)
    logger.info(request.method + ' ' + request.url.path, extra={'extra_info': get_extra_info(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='127.0.0.1', port=8000)

UPDATE 2

In addition to the previous update, you could have a background task for logging the data, instead of waiting for the logging to complete before returning the response. A background task "will run only once the response has been sent" (as per Starlette documentation). You can define a task function to run in the background for writing the log data, as shown below:

def write_log_data(request, response):
    logger.info(request.method + ' ' + request.url.path, extra={'extra_info': get_extra_info(request, response)})

Then, inside the middleware, you can pass the (above) task function to the background tasks, as shown below (credits to this post):

from starlette.background import BackgroundTask

@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

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"
      }
    }
  }
}