Pular para o conteúdo principal

Blog

Como Criar Logs Estruturados e Rastreáveis em Aplicações FastAPI

André Felipe Dias

Comentários

Também disponível em:

English

https://www.flickr.com/photos/gregloby/3763720734/

Mensagens de log são fundamentais no monitoramento de aplicações em produção, mas geralmente essas mensagens são inadequadas. Este artigo mostra como melhorar a geração de mensagem de logs em aplicações FastAPI usando o Loguru como módulo de logging, formatação JSON e a identificação de mensagens de log por meio de IDs de requisição.

Introdução

Os logs são a principal fonte de informação sobre o comportamento da aplicação em produção. São cruciais na identificação de problemas, na rastreabilidade de eventos e no monitoramento do desempenho de aplicações. No entanto, a maioria das aplicações registra logs de forma inadequada. Geralmente, os logs são gerados em formato de texto simples, sem estrutura definida, o que dificulta o processamento e a análise. Também é comum a falta de um elo de ligação entre as mensagens de logs, impossibilitando o rastreamento do processamento de uma requisição específica em uma aplicação web, por exemplo.

O objetivo deste artigo é melhorar a geração de mensagem de logs em aplicações FastAPI. Para isso, implantaremos as seguintes funcionalidades:

  1. Uso do Loguru em substituição do módulo padrão de logging do Python. Loguru é uma boa alternativa porque é mais simples de configurar e usar.
  2. Logs no formato JSON. O formato JSON é fácil de ler e possui uma estrutura bem definida que facilita o processamento. É bem aceito na integração com serviços centralizados de logging e com ferramentas de linha de comando tais como jq.
  3. Associação de toda mensagem de log a um ID de requisição para possibilitar o rastreamento de todo o processamento de uma resposta.
  4. Registro do tempo total de processamento de cada requisição. Essa informação será importante para o monitoramento do desempenho da aplicação.
  5. Registro dos casos de erro e exceções da aplicação no log para auxiliar na depuração de problemas.

Logs em Formato JSON

No loguru, a serialização de mensagem de log em JSON é feita através do parâmetro serialize=True durante a inclusão do handler:

$ python -m venv .venv
$ source .venv/bin/activate
(.venv) $ pip install loguru
(.venv) $ python
Python 3.11.4 (main, Aug 10 2023, 20:08:21) [GCC 12.2.0] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> import sys
>>> from loguru import logger
>>> logger.remove()
>>> logger.add(sys.stderr, serialize=True)
1
>>> logger.info('Hello World', request_id='abcdef')
{"text": "2023-10-16 07:22:49.234 | INFO     | __main__:<module>:1 - Hello World\n", "record": {"elapsed": {"repr": "4:50:14.048257", "seconds": 17414.048257}, "exception": null, "extra": {"request_id": "abcdef"}, "file": {"name": "<stdin>", "path": "<stdin>"}, "function": "<module>", "level": {"icon": "i", "name": "INFO", "no": 20}, "line": 1, "message": "Hello World", "module": "<stdin>", "name": "__main__", "process": {"id": 834569, "name": "MainProcess"}, "thread": {"id": 140172067380096, "name": "MainThread"}, "timestamp": {"repr": "2023-10-16 07:22:49.234603-03:00", "timestamp": 1697451769.234603}}}

Abaixo, temos o mesmo log, mas em uma formatação melhorada:

{
    "text": "2023-10-16 07:22:49.234 | INFO     | __main__:<module>:1 - Hello World\n",
    "record": {
        "elapsed": {
            "repr": "4:50:14.048257",
            "seconds": 17414.048257
        },
        "exception": null,
        "extra": {
            "request_id": "abcdef"
        },
        "file": {
            "name": "<stdin>",
            "path": "<stdin>"
        },
        "function": "<module>",
        "level": {
            "icon": "i",
            "name": "INFO",
            "no": 20
        },
        "line": 1,
        "message": "Hello World",
        "module": "<stdin>",
        "name": "__main__",
        "process": {
            "id": 834569,
            "name": "MainProcess"
        },
        "thread": {
            "id": 140172067380096,
            "name": "MainThread"
        },
        "timestamp": {
            "repr": "2023-10-16 07:22:49.234603-03:00",
            "timestamp": 1697451769.234603
        }
    }
}

Há alguns problemas com essa estrutura:

  1. Há muitos níveis de aninhamento. Isso dificulta a leitura e o processamento dos logs.
  2. Alguns campos não são necessários tais como process e thread.
  3. O campo elapsed não representa o tempo de resposta de uma requisição e sim o tempo desde o início da aplicação.
  4. O campo request_id está no campo extra mas ficaria melhor se estivesse no primeiro nível.
  5. O campo level representa o nível da mensagem através de um dicionário com os campos icon, name e no. Seria melhor se fosse um único campo level com o valor contido em name.
  6. A origem da mensagem de log está apresentado no campo text e também desmembrada em diferentes campos de record. Seria melhor se fosse um único campo source com o valor <file.name>:<function>:<line>, tal qual aparece inicialmente.

Vamos resolver esses problemas através de uma serialização customizada de mensagens de log.

Primeira Versão da Serialização Customizada

A recomendação da documentação do Loguru para customizar a serialização de logs é criar uma função específica de serialização e associá-la a um sink. Isto será feito em um novo módulo de nome logging.py:

import sys
import orjson
from loguru import logger
orjson_options = orjson.OPT_NAIVE_UTC
def serialize(record: dict) -> str:
    subset = {
        'timestamp': record['time'].isoformat(),
        'level': record['level'].name,
        'message': record['message'],
        'source': f'{record["file"].name}:{record["function"]}:{record["line"]}',
    }
    subset.update(record['extra'])
    return orjson.dumps(subset, default=str, options=orjson_options).decode()
def init_loguru() -> None:
    logger.remove()
    logger.add(lambda message: print(serialize(message.record), file=sys.stderr))

A função serialize recebe um dicionário record com os campos da mensagem de log e retorna uma string com a mensagem serializada em JSON.

Os campos extra do registro original são apresentados no primeiro nível da mensagem de log (linha 16).

A conversão para JSON é feita pela biblioteca orjson por ser mais rápida que a biblioteca padrão json (linha 17). A chamada à função orjson.dumps usa str como função default para serializar tipos não suportados. Nem sempre os resultados são os melhores possíveis, mas evita que o log seja interrompido por uma exceção de serialização.

O resultado para o comando logger.info('Hello World', request_id='abcdef') agora é:

{"timestamp": "2023-10-16T11:01:03.223737-03:00", "level": "INFO", "message": "Hello World", "source": "<stdin>:<module>:1", "request_id": "abcdef"}

Esta primeira versão da função serialize corrige os problemas listados anteriormente, mas ainda é possível melhorar um pouco mais a função para que produza uma formatação mais amigável durante o desenvolvimento, com endentação e sintaxe colorida.

Segunda Versão com Endentação e Colorização

A segunda versão da função serialize é:

import sys
import orjson
from loguru import logger
from pygments import highlight
from pygments.formatters import Terminal256Formatter
from pygments.lexers import JsonLexer
from . import config
lexer = JsonLexer()
formatter = Terminal256Formatter(style=config.PYGMENTS_STYLE)
orjson_options = orjson.OPT_NAIVE_UTC
if config.DEBUG:
    orjson_options |= orjson.OPT_INDENT_2
def serialize(record: dict) -> str:
    subset = {
        'timestamp': record['time'].isoformat(),
        'level': record['level'].name,
        'message': record['message'],
        'source': f'{record["file"].name}:{record["function"]}:{record["line"]}',
    }
    subset.update(record['extra'])
    formatted_json = orjson.dumps(subset, default=str, option=orjson_options).decode()
    if config.DEBUG:
        formatted_json = highlight(formatted_json, lexer, formatter)
    return formatted_json
def init_loguru() -> None:
    logger.remove()
    logger.add(lambda message: print(serialize(message.record), file=sys.stderr))

Durante o desenvolvimento, quando a variável de configuração DEBUG é True, a formatação do JSON é melhorada com endentação e sintaxe colorida.

A endentação é feita com a adição da opção OPT_INDENT_2 às opções de serialização do orjson (linha 14 e 15).

A sintaxe colorida é feita pela função highlight da biblioteca Pygments (linha 28). O estilo de formatação é definido através da variável de configuração PYGMENTS_STYLE (linha 12), que pode ser alterada de acordo com a preferência do desenvolvedor criando uma variável de ambiente com o mesmo nome.

A saída para o comando logger.info('Hello World', request_id='abcdef') fica assim durante o desenvolvimento e testes:

{
  "timestamp": "2023-10-16T11:01:03.223737-03:00",
  "level": "INFO",
  "message": "Hello World",
  "source": "<stdin>:<module>:1",
  "request_id": "abcdef"
}

Terceira e Última Versão da Serialização Customizada

Ainda há mais um ajuste a ser feito. Em caso de exceção, a mensagem de log deve conter o traceback da exceção. A formatação do traceback será feita pela biblioteca stackprinter, que mostra um contexto melhorado da exceção com os valores das variáveis próximas:

import sys
import orjson
import stackprinter
from loguru import logger
from pygments import highlight
from pygments.formatters import Terminal256Formatter
from pygments.lexers import JsonLexer
from . import config
lexer = JsonLexer()
formatter = Terminal256Formatter(style=config.PYGMENTS_STYLE)
orjson_options = orjson.OPT_NAIVE_UTC
if config.DEBUG:
    orjson_options |= orjson.OPT_INDENT_2
def serialize(record: dict) -> str:
    subset = {
        'timestamp': record['time'].isoformat(),
        'level': record['level'].name,
        'message': record['message'],
        'source': f'{record["file"].name}:{record["function"]}:{record["line"]}',
    }
    subset.update(record['extra'])
    if record['exception']:
        subset['exception'] = stackprinter.format(record['exception'])
    formatted_json = orjson.dumps(subset, default=str, option=orjson_options).decode()
    if config.DEBUG:
        formatted_json = highlight(formatted_json, lexer, formatter)
    return formatted_json
def init_loguru() -> None:
    logger.remove()
    logger.add(lambda message: print(serialize(message.record), file=sys.stderr))  # type: ignore

Ajustes no Projeto FastAPI Mínimo

O gabarito do projeto FastAPI Mínimo precisa de alguns ajustes para adotar as melhorias apresentadas. Primeiro, orjson, pygments e stackprinter devem ser adicionados à lista de dependências do projeto:

$ poetry add orjson=="*" pygments=="*" stackprinter=="*"

Em segundo lugar, a chamada à init_loguru deve ser feita durante a iniciação da aplicação, em resources.py:

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from loguru import logger
from fastapi import FastAPI
from . import config
from .logging import init_loguru
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator:  # noqa: ARG001
    await startup()
    try:
        yield
    finally:
        await shutdown()
async def startup() -> None:
    init_loguru()
    show_config()
    # insert here calls to connect to database and other services
    logger.info('started...')
...

Com essas alterações, as mensagem de log passam a ser geradas no formato JSON. A próxima etapa é incluir o identificador da requisição em todas as mensagens de log.

Rastreamento de Requisições no Log

O processamento da resposta a uma requisição é registrado no log em várias mensagens diferentes, intercaladas com mensagens de outros processamentos concorrentes. Se gerarmos um identificador único para cada requisição e usarmos esse identificador nas mensagens de log subsequentes, poderemos filtrar e agrupar as mensagens de log relacionadas para rastrear todo o processamento.

Para uma aplicação FastAPI, uma boa maneira de gerar o identificador da requisição é através de um middleware. A implementação do middleware apresentada a seguir cria um ID aleatório para cada requisição e o inclui no contexto do logger para que seja incluído em todas as mensagens de log do mesmo processamento. Além disso, também registra o tempo de processamento de cada requisição.

from collections.abc import Callable
from secrets import token_urlsafe
from time import time
from fastapi import Request, Response
from fastapi.responses import JSONResponse
from hypercorn.logging import AccessLogAtoms
from loguru import logger
from . import config
async def log_request_middleware(request: Request, call_next: Callable) -> Response:
    """
    Uniquely identify each request and logs its processing time.
    """
    start_time = time()
    request_id: str = token_urlsafe(config.REQUEST_ID_LENGTH)
    # keep the same request_id in the context of all subsequent calls to logger
    with logger.contextualize(request_id=request_id):
        response = await call_next(request)
        final_time = time()
        elapsed = final_time - start_time
        response_dict = {
            'status': response.status_code,
            'headers': response.headers.raw,
        }
        atoms = AccessLogAtoms(request, response_dict, final_time)  # type: ignore
        logger.info(
            'log request',
            remote_ip=request.headers.get('x-forwarded-for') or atoms['h'],
            schema=request.headers.get('x-forwarded-proto') or atoms['S'],
            protocol=atoms['H'],
            method=atoms['m'],
            path_with_query=atoms['Uq'],
            status_code=response.status_code,
            response_length=atoms['b'],
            elapsed=elapsed,
            referer=atoms['f'],
            user_agent=atoms['a'],
        )
    response.headers['X-Request-ID'] = request_id
    response.headers['X-Processed-Time'] = str(elapsed)
    return response

Vamos analisar o código nas próximas seções.

Identificador da Requisição

O identificador da requisição é gerado através da função secrets.token_urlsafe, que retorna um texto aleatório codificado em Base64.

O tamanho do identificador não precisa ser muito grande pois não será usado para criptografia. Usando um valor padrão de 8 bytes para a variável de configuração REQUEST_ID_LENGTH, temos 18.446.744.073.709.551.616 combinações possíveis, o que é suficiente para um identificador aleatório.

Repassando o Identificador da Requisição para Chamadas Subsequentes ao Logger

Na linha 21, o identificador da requisição é incluído no contexto do logger através da função contextualize. Dessa forma, todas as chamadas a logger na mesma thread ou tarefa assíncrona recebem automaticamente o identificador da requisição dentro do campo extra, sem a necessidade de passá-lo explicitamente como parâmetro. Esse campo extra é depois expandido para o primeiro nível na função serialize, conforme apresentado na seção anterior.

Mensagem de Log da Requisição

Ao final do processamento da requisição, é apresentada uma mensagem de log com o tempo de processamento e os dados da requisição e da resposta. Ao invés de extrair as informações da requisição e da resposta diretamente dos objetos request e response, é usado um objeto AccessLogAtoms para fazer o trabalho pesado (linha 30).

Registrando o Middleware na Aplicação

O middleware deve ser registrado na aplicação para que seja ativado. No projeto FastAPI mínimo, isso é feito no arquivo main.py:

from fastapi import FastAPI
from fastapi.exceptions import RequestValidationError
from starlette.middleware.base import BaseHTTPMiddleware
from . import config  # noqa: F401
from .resources import lifespan
from .routers import hello
from .middleware import log_request_middleware
app = FastAPI(
    title='fastapi_minimum',
    lifespan=lifespan,
)
routers = (
    hello.router,
)
for router in routers:
    app.include_router(router)
app.add_middleware(BaseHTTPMiddleware, dispatch=log_request_middleware)

A inclusão do middleware é feita através do método add_middleware da aplicação FastAPI, passando a função log_request_middleware como parâmetro dispatch (linha 22).

Mensagem de Log com o Middleware Registrado

Considerando o endpoint /hello do projeto FastAPI mínimo:

from fastapi import APIRouter
from loguru import logger
router = APIRouter()
@router.get('/hello')
async def hello_world() -> dict[str, str]:
    logger.info('Hello world!')
    return {'message': 'Hello World'}

Com o middleware implementado e registrado na aplicação, as mensagens de log para uma chamada ao endpoint /hello ficam assim:

{
  "timestamp": "2023-10-18T07:23:59.274914+00:00",
  "level": "INFO",
  "message": "Hello world!",
  "source": "hello.py:hello_world:9",
  "request_id": "M5L1Dftkkog"
}

{
  "timestamp": "2023-10-18T07:23:59.275833+00:00",
  "level": "INFO",
  "message": "log request",
  "source": "middleware.py:log_request_middleware:31",
  "request_id": "M5L1Dftkkog",
  "remote_ip": "192.168.112.3:47140",
  "schema": "http",
  "protocol": "2",
  "method": "GET",
  "path_with_query": "/hello",
  "status_code": 200,
  "response_length": "25",
  "elapsed": 0.0012085437774658203,
  "referer": "-",
  "user_agent": "HTTPie/3.2.2"
}

A primeira mensagem de log é gerada na chamada ao endpoint /hello, conforme indicado pelo campo source. A segunda é a mensagem de log final do processamento, gerada no log_request_middleware.

Registrando Erros de Validação no Log

Um erro de validação 422 é gerado quando o endpoint recebe parâmetros inválidos, que não podem ser processados. Seja o endpoint abaixo que calcula a divisão de dois números inteiros:

@router.get('/divide')
def divide(a: int, b: int) -> float:
    return a / b

Chamar esse endpoint com um float no lugar de um int gera um erro de validação:

$ httpx --http2 --no-verify 'https://localhost/api/divide?a=1.5&b=2'
HTTP/2 422 Unprocessable Entity
alt-svc: h3=":443"; ma=2592000
content-type: application/json
date: Thu, 19 Oct 2023 08:05:14 GMT
server: Caddy
server: hypercorn-h2
x-processed-time: 0.0010986328125
x-request-id: Gw5vFSCiHFM
content-length: 201

{
    "detail": [
        {
            "type": "int_parsing",
            "loc": [
                "query",
                "a"
            ],
            "msg": "Input should be a valid integer, unable to parse string as an
integer",
            "input": "1.5",
            "url": "https://errors.pydantic.dev/2.4/v/int_parsing"
        }
    ]
}

Mesmo com o middleware de log de requisição implementado, o erro de validação não é registrado no log. Temos apenas a mensagem de log final do processamento da requisição:

{
  "timestamp": "2023-10-19T08:05:14.601434+00:00",
  "level": "INFO",
  "message": "log request",
  "source": "middleware.py:log_request_middleware:30",
  "request_id": "Gw5vFSCiHFM",
  "remote_ip": "192.168.128.3:37830",
  "schema": "http",
  "protocol": "2",
  "method": "GET",
  "path_with_query": "/divide?a=1.5&b=2",
  "status_code": 422,
  "response_length": "201",
  "elapsed": 0.0010986328125,
  "referer": "-",
  "user_agent": "python-httpx/0.24.1"
}

Registrar erros de validação no log pode ser útil para depurar problemas de validação durante o desenvolvimento e testes. Isso será feito através de um exception handler que interceptará as exceções do tipo RequestValidationError. A ideia é apenas registrar o erro de validação no log e depois chamar o exception handler padrão do FastAPI :

from fastapi import Request
from fastapi.exception_handlers import (
    request_validation_exception_handler as _request_validation_exception_handler,
)
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from loguru import logger
async def request_validation_exception_handler(
    request: Request, exc: RequestValidationError
) -> JSONResponse:
    """
    This is a wrapper to the default RequestValidationException handler of FastAPI
    that logs the exception for easier debugging.
    """
    method = request.get('method', 'GET')
    query_string = request['query_string'].decode()
    path_with_query = request['path'] + ('?' + query_string if query_string else '')
    logger.info('request validation exception', method=method, path_with_query=path_with_query, detail=exc.errors())
    return await _request_validation_exception_handler(request, exc)

Uma chamada a /divide?a=1.5&b=2 agora gera a seguinte mensagem de log:

{
  "timestamp": "2023-10-19T08:05:14.600671+00:00",
  "level": "INFO",
  "message": "request validation exception",
  "source": "exception_handlers.py:request_validation_exception_handler:17",
  "request_id": "Gw5vFSCiHFM",
  "method": "GET",
  "path_with_query": "/divide?a=1.5&b=2",
  "detail": [
    {
      "type": "int_parsing",
      "loc": [
        "query",
        "a"
      ],
      "msg": "Input should be a valid integer, unable to parse string as an integer",
      "input": "1.5",
      "url": "https://errors.pydantic.dev/2.4/v/int_parsing"
    }
  ]
}

Registrando Exceções Não Tratadas no Log

Um outro tipo de erro que pode ocorrer durante o processamento de uma requisição é uma exceção não tratada. Por exemplo, se o endpoint /divide for chamado com o parâmetro b=0, será gerada uma exceção ZeroDivisionError. O cliente recebe uma resposta 500 com uma mensagem de erro genérica:

$ httpx --http2 --no-verify 'https://localhost/api/divide?a=1&b=0'
HTTP/2 500 Internal Server Error
alt-svc: h3=":443"; ma=2592000
content-type: text/plain; charset=utf-8
date: Thu, 19 Oct 2023 20:58:55 GMT
server: Caddy
server: hypercorn-h2
content-length: 21

Internal Server Error

Na configuração atual, a aplicação FastAPI não registra a exceção no log, mas a exceção é repassada para o servidor ASGI, que a cria uma mensagem de log do servidor:

[2023-10-19 20:58:53 +0000] [10] [INFO] Running on http://0.0.0.0:5000 (CTRL + C to quit)
[2023-10-19 20:58:55 +0000] [10] [ERROR] Error in ASGI Framework
Traceback (most recent call last):
  File "/venv/lib/python3.11/site-packages/starlette/middleware/base.py", line 108, in __call__
    response = await self.dispatch_func(request, call_next)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/app/fastapi_minimum/middleware.py", line 22, in log_request_middleware
    response = await call_next(request)
               ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/venv/lib/python3.11/site-packages/starlette/middleware/base.py", line 84, in call_next
    raise app_exc
  File "/venv/lib/python3.11/site-packages/starlette/middleware/base.py", line 70, in coro
    await self.app(scope, receive_or_disconnect, send_no_error)
  File "/venv/lib/python3.11/site-packages/starlette/middleware/exceptions.py", line 79, in __call__
    raise exc
  File "/venv/lib/python3.11/site-packages/starlette/middleware/exceptions.py", line 68, in __call__
    await self.app(scope, receive, sender)
  File "/venv/lib/python3.11/site-packages/fastapi/middleware/asyncexitstack.py", line 20, in __call__
    raise e
  File "/venv/lib/python3.11/site-packages/fastapi/middleware/asyncexitstack.py", line 17, in __call__
    await self.app(scope, receive, send)
  File "/venv/lib/python3.11/site-packages/starlette/routing.py", line 718, in __call__
    await route.handle(scope, receive, send)
  File "/venv/lib/python3.11/site-packages/starlette/routing.py", line 276, in handle
    await self.app(scope, receive, send)
  File "/venv/lib/python3.11/site-packages/starlette/routing.py", line 66, in app
    response = await func(request)
               ^^^^^^^^^^^^^^^^^^^
  File "/venv/lib/python3.11/site-packages/fastapi/routing.py", line 273, in app
    raw_response = await run_endpoint_function(
                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/venv/lib/python3.11/site-packages/fastapi/routing.py", line 190, in run_endpoint_function
    return await dependant.call(**values)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/app/fastapi_minimum/routers/hello.py", line 15, in divide
    return a / b
           ~~^~~
ZeroDivisionError: division by zero
...

O problema é que não há conexão entre a mensagem de log e a requisição, tornando difícil rastrear a requisição. O ideal é que a mensagem de log da exceção contenha o identificador da requisição tal como nas mensagens de log anteriores.

Normalmente, criaríamos um exception handler para registrar no log as exceções não tratadas:

async def unhandled_exception_handler(request: Request, exc: Exception) -> JSONResponse:
    method = request.get('method', 'GET')
    query_string = request['query_string'].decode()
    path_with_qs = request['path'] + ('?' + query_string if query_string else '')
    logger.opt(exception=exc).info('Unhandled exception', method=method, path=path)
    return JSONResponse(content={'detail': 'Internal Server Error'}, status_code=500)

O exception handler acima cria a mensagem de log com o identificador da requisição e a exceção, mas não evita que a exceção seja repassada para o servidor ASGI, que continua gerando uma mensagem de log duplicada e descontextualizada (veja esse link para mais detalhes).

Para evitar isso, vamos capturar exceções não tratadas no próprio middleware de log de requisição:

async def log_request_middleware(request: Request, call_next: Callable) -> Response:
    """
    Uniquely identify each request and logs its processing time.
    """
    start_time = time()
    request_id: str = token_urlsafe(config.REQUEST_ID_LENGTH)
    exception = None
    # keep the same request_id in the context of all subsequent calls to logger
    with logger.contextualize(request_id=request_id):
        try:
            response = await call_next(request)
        except Exception as exc:
            exception = exc
            response = PlainTextResponse('Internal Server Error', status_code=500)
        final_time = time()
        elapsed = final_time - start_time
        response_dict = {
            'status': response.status_code,
            'headers': response.headers.raw,
        }
        atoms = AccessLogAtoms(request, response_dict, final_time)  # type: ignore
        data = {
            'remote_ip': request.headers.get('x-forwarded-for') or atoms['h'],
            'schema': request.headers.get('x-forwarded-proto') or atoms['S'],
            'protocol': atoms['H'],
            'method': atoms['m'],
            'path_with_query': atoms['Uq'],
            'status_code': response.status_code,
            'response_length': response_length,
            'elapsed': elapsed,
            'referer': atoms['f'],
            'user_agent': atoms['a'],
        }
        if not exception:
            logger.info('log request', **data)
        else:
            logger.opt(exception=exception).error('unhandled exception', **data)
    response.headers['X-Request-ID'] = request_id
    response.headers['X-Processed-Time'] = str(elapsed)
    return response

Exceções não-tratadas são capturadas no bloco try..except (linha 23 a 27) e a exceção fica armazenada na variável exception para ser usada posteriormente na mensagem de log. A resposta padrão para exceções não-tratadas é uma resposta 500 com uma mensagem de erro genérica (linha 27).

O tipo da mensagem de log da requisição vai depender se houve exceção ou não. Se não houve exceção, a mensagem de log é gerada com o nível INFO (linha 48). Se houve exceção, a mensagem de log é gerada com o nível ERROR (linha 50) e a exceção incluída através do método opt, que permite alterar a configuração de uma chamada específica ao logger.

O traceback da exceção não-tratada é apresentada junto com a mensagem de log da requisição no campo exception:

{
  "timestamp": "2023-10-20T21:18:55.045870+00:00",
  "level": "ERROR",
  "message": "Unhandled exception",
  "source": "middleware.py:log_request_middleware:50",
  "request_id": "zv9ueo-i1Hg",
  "remote_ip": "192.168.144.3:36196",
  "schema": "http",
  "protocol": "2",
  "method": "GET",
  "path_with_query": "/divide?a=1&b=0",
  "status_code": 500,
  "response_length": "34",
  "elapsed": 0.0005097389221191406,
  "referer": "-",
  "user_agent": "python-httpx/0.24.1",
  "exception": "File \"/venv/lib/python3.11/site-packages/starlette/middleware/base.py\", line 78, in call_next\n    31   async def call_next(request: Request) -> Response:\n (...)\n    74       task_group.start_soon(close_recv_stream_on_response_sent)\n    75       task_group.start_soon(coro)..."
}

Testes Automatizados

Para garantir que as mensagens de log estão sendo geradas corretamente, vamos criar alguns testes automatizados em test_logging.py.

Fixture de Aplicação para Testes

A primeira parte do arquivo test_logging.py é apresentada abaixo:

import json
from collections.abc import AsyncIterable
from unittest.mock import patch
from fastapi import APIRouter, FastAPI, status
from httpx import AsyncClient
from pytest import CaptureFixture, fixture
basic_log_fields = {
    'timestamp',
    'level',
    'message',
    'source',
    'request_id',
    'method',
    'path_with_query',
}
failed_validation_log_fields = basic_log_fields | {'detail'}
request_log_fields = basic_log_fields | {'protocol', 'schema', 'elapsed', 'status_code'}
exception_log_fields = request_log_fields | {'exception'}
@fixture(scope='module')
async def logging_client() -> AsyncIterable[AsyncClient]:
    """
    Independent client/app instance for testing the logging.
    """
    from fastapi_minimum.logging import init_loguru
    from fastapi_minimum.main import (
        BaseHTTPMiddleware,
        RequestValidationError,
        log_request_middleware,
        request_validation_exception_handler,
    )
    router = APIRouter()
    @router.get('/info')
    async def info() -> dict[str, int]:
        return {'data': 1234}
    @router.get('/divide')
    async def divide(a: int, b: int) -> float:
        return a / b
    app = FastAPI()
    app.include_router(router)
    app.add_middleware(BaseHTTPMiddleware, dispatch=log_request_middleware)
    app.add_exception_handler(RequestValidationError, request_validation_exception_handler)
    init_loguru()
    async with AsyncClient(app=app, base_url='http://test_logging') as client:
        yield client
    return

As linhas 8 a 19 definem os campos que devem estar presentes em cada tipo de mensagem de log. Esses campos serão usados nos testes para verificar se as mensagens de log estão sendo geradas corretamente.

A função logging_client (linha 23 a 56) é uma fixture que cria uma nova aplicação com os endpoints necessários para os testes, adiciona o middleware e o exception handler de logging e retorna um cliente HTTP para essa aplicação.

O endpoint /info retorna uma resposta simples (linhas 38 a 40). Representa uma chamada normal a um endpoint sem exceções ou erros de validação.

/divide recebe dois parâmetros inteiros a e b e retorna o resultado da divisão de a por b (linhas 42 a 44). Esse endpoint será usado tanto para testar erros de validação, recebendo um float ao invés de um inteiro, quanto para exceções não tratadas, ao receber b=0 e gerar uma exceção ZeroDivisionError.

A função init_loguru (linha 51) inicializa o logger com a serialização customizada. Pode haver outras iniciações no lifespan da aplicação principal, mas apenas essa é necessária para os testes de logging.

Teste de Formato JSON de Mensagens de Log

Na segunda parte do mesmo arquivo, temos o primeiro teste que verifica se a mensagem de log é gerada no formato JSON e se contém os campos esperados:

async def test_json_logging(
    logging_client: AsyncClient, capsys: CaptureFixture
) -> None:
    """
    Test that the log is in JSON format.
    """
    with patch(
        'fastapi_minimum.logging.highlight', side_effect=lambda x, y, z: x
    ):  # prevents highlighting
        response = await logging_client.get('/info')
    assert response.status_code == status.HTTP_200_OK
    assert response.json() == {'data': 1234}
    log = json.loads(capsys.readouterr().err)
    assert request_log_fields <= set(log.keys())
    assert log['level'] == 'INFO'
    assert 'exception' not in log

capsys é uma fixture do pytest que captura a saída padrão e de erro.

As linhas 65 a 67 substituem a função highlight no módulo logging por uma função que retorna a entrada sem alterações. Isso evita que a saída para o terminal seja colorida durante o teste, o que dificultaria a comparação com o JSON esperado.

A mensagem de log deve conter pelo menos os campos de request_log_fields (linha 73).

Como não é uma exceção, o nível deve ser INFO (linha 74) e o campo exception não deve estar presente (linha 75).

Teste de Mensagem de Log para Erro de Validação

Na terceira parte de test_logging.py, temos o teste que verifica se a mensagem de log é gerada corretamente para um erro de validação:

async def test_logging_422_exception(
    app_with_logging_routes: FastAPI, client: AsyncClient, capsys: CaptureFixture
) -> None:
    """
    Test if the log contains the exception when the request is invalid.
    """
    with patch(
        'fastapi_minimum.logging.highlight', side_effect=lambda x, y, z: x
    ):  # prevents highlighting
        response = await client.get('/test_logging/divide', params={'a': 1.1, 'b': 0})
    assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
    detail = response.json()['detail']
    # there must be 2 log entries: 1 for the exception and 1 for the request
    text = capsys.readouterr().err
    position = text.index('\n}\n{\n')
    # test validation log
    validation_log = json.loads(text[: position + 2])
    assert failed_validation_log_fields <= set(validation_log.keys())
    assert 'exception' not in validation_log
    assert validation_log['level'] == 'INFO'
    assert detail == validation_log['detail']
    # test request_log
    request_log = json.loads(text[position + 3 :])
    assert request_log_fields <= set(request_log.keys())
    assert request_log['level'] == 'INFO'
    assert 'exception' not in request_log

A resposta à requisição que provoca o erro de validação gera duas mensagem de log: uma para a exceção e outra para a requisição. As mensagens estão separadas pela sequência de caracteres \n}\n{\n (linha 93).

A primeira mensagem é a da exceção. Uma das condições do teste é que o campo detail da mensagem de log seja igual ao campo detail da resposta da requisição.

A segunda mensagem é a da requisição. A verificação dos campos é a mesma do teste anterior.

Teste de Mensagem de Log para Exceção Não Tratada

Na quarta parte de test_logging.py, temos o teste que verifica se a mensagem de log é gerada corretamente para uma exceção não tratada:

async def test_logging_500_exception(
    app_with_logging_routes: FastAPI, client: AsyncClient, capsys: CaptureFixture
) -> None:
    """
    Test the log message of a unhandled exception.
    """
    with patch(
        'fastapi_minimum.logging.highlight', side_effect=lambda x, y, z: x
    ):  # prevents highlighting
        response = await client.get('/test_logging/divide', params={'a': 1, 'b': 0})
    assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
    assert response.text == 'Internal Server Error'
    log = json.loads(capsys.readouterr().err)
    assert exception_log_fields <= set(log.keys())
    assert log['level'] == 'ERROR'

A resposta da requisição é uma resposta 500 com a mensagem de erro genérica (linhas 119 e 120). A mensagem de log deve conter pelo menos os campos de exception_log_fields (linha 123) e o nível de log deve ser ERROR (linha 124).

Teste de Tipos Não Serializáveis no Log

Na quinta e última parte de test_logging.py, temos o teste que verifica se tipos que não são serializáveis não causam erros no log:

async def test_default_encoding(logging_client: AsyncClient, capsys: CaptureFixture) -> None:
    """
    Test the encoding behavior for non-serializable types.
    """
    from datetime import datetime
    from pathlib import Path
    from loguru import logger
    # Path, datetime and set are non-serializable types by default
    for param in (Path('.'), datetime.now(), {1, 2}):
        logger.info('test param encoding', param=param)
        assert 'TypeError:' not in capsys.readouterr().err

Apesar de não serem serializáveis para JSON por padrão, os tipos Path, datetime e set não devem causar erros no log porque são convertidos para str antes de serem serializados, conforme definido na função serialize em logging.py.

Conclusão

FastAPI é um dos frameworks mais produtivos e eficientes para desenvolver REST APIs em Python. No entanto, o registro de mensagens de log não faz parte das funcionalidades nativas do FastAPI e precisa ser implementado à parte.

A solução apresentada neste artigo começa com a configuração do Loguru para gerar mensagens de log no formato JSON. A partir daí, criamos configurações adicionais ao FastAPI baseadas em middleware e um exception handler para gerar mensagens de log com identificadores únicos para cada requisição, registrar o tempo de processamento de cada resposta e registrar erros de validação e exceções não tratadas. Essas configurações foram testadas e validadas através de testes automatizados.

O resultado final é um sistema de log eficiente e flexível, que pode ser usado como base para o desenvolvimento de qualquer aplicação FastAPI. Ajustes podem ser feitos para atender às necessidades específicas de cada projeto, mas as necessidades básicas de monitoramento e depuração já estão cobertas.

O código apresentado pode ser usado com poucas modificações em outros projetos FastAPI. Mas se você quiser começar um novo projeto FastAPI com essas funcionalidades, a melhor opção é usar o gabarito disponibilizado no GitHub. Para instanciar um novo projeto, é necessário usar o cookiecutter. Recomendo combiná-lo com pipx:

$ pipx run cookiecutter gh:andredias/perfect_python_project \
       -c fastapi-minimum

Referências

Próximo artigo: Desenvolvendo uma Aplicação em FastAPI para Compartilhamento de Arquivos - Parte 1

Artigo anterior: Empacotamento e Distribuição do Projeto FastAPI Mínimo

Comentários

Comments powered by Disqus