在 FastAPI 应用程序中监控和日志记录:在 FastAPI 中设置日志记录

有效的监控和日志记录对于维护和调试 FastAPI 应用程序至关重要。日志记录有助于跟踪应用程序行为、识别问题和了解使用模式。本博客将指导您在 FastAPI 中设置日志记录并演示如何有效地使用它。

日志记录为何重要

日志记录对于以下方面至关重要：

调试：快速识别和修复问题。

监控：跟踪应用程序性能和用户行为。

审计：记录重要事件以确保安全性和合规性。

分析：了解趋势并做出数据驱动的决策。

在 FastAPI 中设置日志记录

FastAPI 与 Python 的标准日志记录模块无缝集成。让我们逐步设置基本日志记录配置。

步骤 1：导入日志记录模块

首先，在 main.py 文件中导入日志记录模块：

import logging
from fastapi import FastAPI


app = FastAPI()

步骤 2：配置日志记录

根据您的需要配置日志记录设置。以下是一个简单配置的示例：

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")


logger = logging.getLogger(__name__)

步骤 3：记录消息

使用记录器记录各种严重性级别的消息：

@app.get("/")
def read_root():
    logger.info("Root endpoint was called")
    return {"message": "Hello World"}


@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
    logger.debug(f"Fetching item with ID: {item_id}")
    if q:
        logger.warning(f"Query parameter provided: {q}")
    return {"item_id": item_id, "q": q}

步骤4：运行应用程序

运行您的 FastAPI 应用程序并访问端点以查看日志的运行情况：

uvicorn main:app --reload

自定义日志配置

您可以自定义日志配置以满足您的要求，例如将日志写入文件、设置不同的日志级别或添加处理程序。以下是更高级配置的示例：

import logging
import sys


# Create custom logger
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)


# Create handlers
c_handler = logging.StreamHandler(sys.stdout)
f_handler = logging.FileHandler('app.log')
c_handler.setLevel(logging.WARNING)
f_handler.setLevel(logging.DEBUG)


# Create formatters and add them to handlers
c_format = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
f_format = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
c_handler.setFormatter(c_format)
f_handler.setFormatter(f_format)


# Add handlers to the logger
logger.addHandler(c_handler)
logger.addHandler(f_handler)

使用日志中间件

您可以创建中间件来记录请求和响应。这有助于监控 API 调用和响应以进行调试和分析。

from fastapi import Request


@app.middleware("http")
async def log_requests(request: Request, call_next):
    logger.info(f"Request: {request.method} {request.url}")
    response = await call_next(request)
    logger.info(f"Response: {response.status_code}")
    return response

常用的演示

演示 1：记录用户身份验证

@app.post("/login")
def login(username: str, password: str):
    logger.info(f"Login attempt for user: {username}")
    if username == "admin" and password == "secret":
        logger.info("Login successful")
        return {"message": "Login successful"}
    else:
        logger.error("Login failed")
        return {"message": "Login failed"}, 401

演示 2：记录数据库操作

@app.get("/users/{user_id}")
def get_user(user_id: int):
    logger.debug(f"Fetching user with ID: {user_id}")
    user = {"id": user_id, "name": "John Doe"}  # Simulated database fetch
    logger.info(f"User fetched: {user}")
    return user

演示 3：使用不同日志级别进行日志记录

此示例演示如何在 FastAPI 中以不同级别（DEBUG、INFO、WARNING、ERROR 和 CRITICAL）记录消息。

@app.get("/log_levels")
def log_levels():
    logger.debug("This is a DEBUG message")
    logger.info("This is an INFO message")
    logger.warning("This is a WARNING message")
    logger.error("This is an ERROR message")
    logger.critical("This is a CRITICAL message")
    return {"message": "Logged messages at different levels"}

演示 4：记录异常

此示例展示如何记录 FastAPI 应用程序中发生的异常。

@app.get("/cause_exception")
def cause_exception():
    try:
        raise ValueError("This is a simulated exception")
    except ValueError as e:
        logger.exception("An error occurred: %s", e)
        return {"message": "An error occurred"}, 500

演示 5：使用 JSON 进行结构化日志记录

使用 JSON 进行结构化日志记录有利于更轻松地解析和分析日志。以下是设置方法。

import json_log_formatter


formatter = json_log_formatter.JSONFormatter()


json_handler = logging.FileHandler(filename='app.json')
json_handler.setFormatter(formatter)


logger.addHandler(json_handler)


@app.get("/structured_logging")
def structured_logging():
    logger.info("Structured logging with JSON", extra={"custom_key": "custom_value"})
    return {"message": "Logged a structured message"}

演示 6：使用关联 ID 记录请求和响应

向日志消息添加关联 ID 可以帮助跟踪通过应用程序的请求。

import uuid


@app.middleware("http")
async def add_correlation_id(request: Request, call_next):
    correlation_id = str(uuid.uuid4())
    logger.info(f"Request ID: {correlation_id} - {request.method} {request.url}")
    response = await call_next(request)
    response.headers["X-Correlation-ID"] = correlation_id
    logger.info(f"Response ID: {correlation_id} - {response.status_code}")
    return response


@app.get("/correlation_id")
def get_correlation_id():
    return {"message": "Check logs for correlation ID"}

演示 7：使用 SQLAlchemy 记录 SQL 查询

如果您正在使用 SQLAlchemy，则可以启用 SQL 查询的日志记录以监控数据库交互。

from sqlalchemy import create_engine, event


DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL, echo=True)


@event.listens_for(engine, "before_cursor_execute")
def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    logger.debug(f"SQL: {statement} - Params: {parameters}")


@app.get("/log_sql")
def log_sql():
    with engine.connect() as connection:
        result = connection.execute("SELECT 1")
        return {"result": result.fetchone()}

演示 8：异步日志记录

使用异步端点时，确保正确处理日志记录非常重要。

import asyncio


@app.get("/async_logging")
async def async_logging():
    await asyncio.sleep(1)
    logger.info("Asynchronous logging after delay")
    return {"message": "Logged asynchronously"}

演示 9：使用外部日志记录服务

您可以配置日志记录以将日志发送到外部服务，如 Logstash、Fluentd 或基于云的日志记录解决方案。以下是使用logging.handlers.HTTPHandler将日志发送到HTTP端点的示例：

import logging.handlers


http_handler = logging.handlers.HTTPHandler(
    'localhost:5000', 
    '/log', 
    method='POST',
    secure=False,
)


logger.addHandler(http_handler)


@app.get("/external_logging")
def external_logging():
    logger.info("This log will be sent to an external HTTP endpoint")
    return {"message": "Logged to external service"}

演示 10：轮换日志文件

为了防止日志文件无限增长，您可以使用轮换日志文件。

from logging.handlers import RotatingFileHandler


rotating_handler = RotatingFileHandler('app_rotating.log', maxBytes=2000, backupCount=5)
rotating_handler.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s'))


logger.addHandler(rotating_handler)


@app.get("/rotating_logging")
def rotating_logging():
    for i in range(100):
        logger.info(f"Rotating log entry {i}")
    return {"message": "Logged with rotating file handler"}

这些演示应该可以帮助您了解在FastAPI应用程序中设置和使用日志记录的各个方面。您可以混合搭配这些技术来创建适合您特定需求的全面日志记录策略。有效的日志记录对于监控和维护 FastAPI 应用程序至关重要。通过设置强大的日志记录配置，您可以深入了解应用程序的行为，快速识别和解决问题，并确保运行顺畅。本博客中提供的示例演示了如何在 FastAPI 中实现日志记录，从而更轻松地跟踪和调试应用程序的活动。

写在最后：

1. 良好的编码习惯，不仅仅是逻辑清楚，命名规范和注释充分，也包含合理的日志记录，所以，产线的日志记录就格外的重要了。只有两者并存，才能最大化发挥日志的作用。本篇博文，详细地介绍了各种日志记录方式，期待您实践后的反馈
2. 本篇博文中介绍了FastAPI的日志记录，但是，往往一个完整的系统，不仅仅是应用层，还包含网络代理，消息中间件和数据持久化（数据库等）。它们也有各自的日志，我们也要关注这些环节，才能完美的分析出问题的所在