背景简介
在现有项目中添加 Log 模块，供其他模块使用。
前置信息
- Python 3.11.13 【Conda - 创建 Python 环境】
详细信息
文件结构
.
├── app
│ ├── api
│ │ └── v1
│ │ ├── health.py
│ │ └── redis.py
│ ├── core
│ │ ├── config.py # logging 环境变量配置模块
│ │ ├── database.py
│ │ ├── __init__.py
│ │ ├── logging.py # logging set up 实现模块
│ │ └── redis.py
│ ├── models
│ │ ├── __init__.py
│ │ └── user_basic_info.py
│ ├── __init__.py
│ └── main.py # logging 跟随系统初始化配置
├── tests
│ ├── test_health.py
│ └── test_redis.py
├── alembic.ini
├── .env
└── requirements.txt
代码准备
- 更新/新建
app/core/config.py
# ... other imports
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
# ... your existing settings ...
# Logging Configuration
LOG_LEVEL: str = "INFO"
LOG_FORMAT: str = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
# ... other settings ...
class Config:
env_file = ".env"
case_sensitive = True
settings = Settings()
- 更新
.env
LOG_LEVEL=DEBUG
- 新建
app/core/logging.py
"""Centralized logging configuration for the application.
This module provides a `setup_logging` function to configure the root logger
with a consistent format and level. It can be imported and called at the
start of any application component (e.g., FastAPI, Celery worker) to ensure
uniform logging.
The module can also be executed directly to test the logging configuration.
"""
import logging
import sys
from typing import Optional
from app.core.config import settings
def setup_logging(
level: Optional[str] = None,
format_string: Optional[str] = None,
) -> None:
"""Configures the root logger for the application.
This function sets up the basic configuration for the logging module,
including the log level, format, and the handler. It is designed to be
called once at the application's startup.
Args:
level: The logging level (e.g., "INFO", "DEBUG"). If None, it will
be read from the application's settings.
format_string: The log message format string. If None, it will be
read from the application's settings.
"""
# Determine the effective log level
log_level = (level or settings.LOG_LEVEL).upper()
log_format = format_string or settings.LOG_FORMAT
# Create a formatter
formatter = logging.Formatter(log_format)
# Create a stream handler (writes to sys.stderr by default)
stream_handler = logging.StreamHandler(sys.stderr)
stream_handler.setFormatter(formatter)
# Configure the root logger
root_logger = logging.getLogger()
root_logger.setLevel(log_level)
# Avoid adding multiple handlers if setup_logging is called multiple times
if not root_logger.handlers:
root_logger.addHandler(stream_handler)
logging.info(
"Logging configured with level '%s' and format '%s'.",
log_level,
log_format,
)
if __name__ == "__main__":
"""Standalone execution for testing the logging configuration.
When this script is run directly, it configures the logger and emits
messages at various levels to demonstrate the output.
"""
print("--- Testing Logging Configuration ---")
setup_logging() # Use default settings from config.py
# Get a logger for this module
logger = logging.getLogger(__name__)
# Log messages at different severity 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.")
print("--- Test Complete ---")
验证
- 调用 logging 模块
python -m app.core.logging
--- Testing Logging Configuration ---
2025-10-28 23:01:05,156 - root - INFO - Logging configured with level 'DEBUG' and format '%(asctime)s - %(name)s - %(levelname)s - %(message)s'.
2025-10-28 23:01:05,156 - __main__ - DEBUG - This is a DEBUG message.
2025-10-28 23:01:05,156 - __main__ - INFO - This is an INFO message.
2025-10-28 23:01:05,156 - __main__ - WARNING - This is a WARNING message.
2025-10-28 23:01:05,156 - __main__ - ERROR - This is an ERROR message.
2025-10-28 23:01:05,156 - __main__ - CRITICAL - This is a CRITICAL message.
--- Test Complete ---
推荐配置
- 在 main.py 中使用
lifespan配置项目启动时直接初始化 logging 模块。
"""FastAPI application entry point.
This module defines the main FastAPI application instance, including
lifecycle management for resources like the Redis client.
"""
import logging
from app.core import setup_logging
from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.api.v1 import api_router
from app.core.config import settings
from redis.asyncio import Redis as AsyncRedis
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Manages application lifecycle events.
This context manager handles the startup and shutdown logic for the
FastAPI application.
It's designed to be test-friendly: if a Redis client is already
present in `app.state` (e.g., a mock), it will not be overwritten.
Args:
app: The FastAPI application instance.
"""
# Initialize logging and other startup tasks.
setup_logging()
# --- Startup ---
logging.info("Application startup...")
# Only create a Redis client if one doesn't already exist.
# This allows tests to inject a mock client before the app starts.
if not hasattr(app.state, "redis") or app.state.redis is None:
logging.info("Creating and connecting real Redis client...")
app.state.redis = AsyncRedis.from_url(
settings.REDIS_URL,
encoding="utf-8",
decode_responses=True,
)
await app.state.redis.ping()
logging.info("Real Redis client connected and initialized.")
else:
logging.info("Using existing Redis client (likely a mock for testing).")
yield # Application runs here, waiting for requests.
# --- Shutdown ---
logging.info("Application shutdown...")
# Close the Redis connection if it exists.
if hasattr(app.state, "redis") and app.state.redis:
await app.state.redis.close()
logging.info("Redis client closed.")
def create_app() -> FastAPI:
"""Create and configure the FastAPI application.
This function factory creates the FastAPI app, sets up metadata,
includes routers, and configures the lifespan manager.
Returns:
The configured FastAPI application instance.
"""
app = FastAPI(
title=settings.APP_NAME,
version=settings.VERSION,
debug=settings.DEBUG,
description="Standard FastAPI Structure.",
lifespan=lifespan,
)
# Register routers
app.include_router(api_router, prefix=settings.API_V1_STR)
return app
app = create_app()
以上便是本文的全部内容，感谢您的阅读，如遇到任何问题，欢迎在评论区留言讨论。