简单描述

• fastapi的简单实战，且用uvicorn将日志同时输出到控制台和日志文件中

main.py

import signal
import sys
from contextlib import asynccontextmanagerfrom fastapi import FastAPI
import uvicorn
from fastapi.staticfiles import StaticFilesfrom settings import settings
from routers.xxx import xxx_router
from common.logging import logger@asynccontextmanager
async def lifespan(app: FastAPI):# Code to run on startuplogger.info("Starting up...")# You can initialize resources here (e.g., database connections)yield# Code to run on shutdownlogger.info("Shutting down...")# Clean up resources hereapp = FastAPI(lifespan=lifespan,debug=settings.debug,title="xxx平台",description='xxx',version='1.0.0',docs_url='/docs',redoc_url='/redoc',)app.mount("/static", StaticFiles(directory=settings.static_dir), name="static")app.include_router(xxx_router, prefix="/api/xxx", tags=["xxx"])@app.get("/")
def read_root():return {"data": "welcome to xxx center"}def signal_handler(sig, frame):print(f'Received Ctrl+C! {sig} exiting...')# 在这里执行任何必要的清理工作sys.exit()def run_http():signal.signal(signal.SIGINT, signal_handler)signal.signal(signal.SIGTERM, signal_handler)uvicorn.run(app="main:app", host=settings.server_ip, port=settings.server_port,log_config="uvicorn_config.json",log_level=settings.log_level, workers=settings.workers)if __name__ == "__main__":run_http()

settings/settings.py

import json
import os
import sys# 获取可执行文件所在的目录路径
if getattr(sys, 'frozen', False):# 如果程序是被打包成了单一文件，这个条件是TrueBASE_DIR = os.path.dirname(sys.executable)
else:# 如果程序是直接运行的.py文件，这个条件是TrueBASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))class BaseConfig(object):def __init__(self):self.config_path = self.get_config_path()self.config = self.get_config()@staticmethoddef get_config_path():config_path = os.path.join(BASE_DIR, 'config.json')if os.path.exists(config_path):return config_pathprint(f'config file {config_path} not exists')def get_config(self):try:with open(self.config_path, 'r', encoding="utf-8") as f:return json.load(f)except FileNotFoundError:print(f'config file {self.config_path} not exists')except json.decoder.JSONDecodeError:print(f'config file {self.config_path} format error')class ProjectConfig(BaseConfig):"""项目中的配置信息"""def get_server(self):"""后台服务配置信息"""try:return self.config["server"]["ip"], self.config["server"]["port"]except Exception as e:print(f'http server section not exists: {e}')def get_debug_mode(self):"""是否开启debug模式"""try:return True if self.config.get("debug_mode") == "true" else Falseexcept Exception as e:print(f'debug_mode section not exists: {e}')def get_workers(self):"""开启进程数"""try:return self.config["workers"]except Exception as e:print(f'process workers section not exists: {e}')def get_log(self):"""日志配置信息"""try:return self.config["logfile"]["level"], self.config["logfile"]["dir"], self.config["logfile"]["max_age"]except Exception as e:print(f'log file section not exists: {e}')def get_mysql(self):"""mysql数据库配置信息"""try:return self.config["mysql"]["default"]except Exception as e:print(f'mysql section not exists: {e}')def get_identify(self):"""识别服务配置信息"""try:return self.config["identify"]["ip"], self.config["identify"]["port"], self.config["identify"]["api_path"]except Exception as e:print(f'identify server section not exists: {e}')def get_white_list(self):"""请求白名单列表"""try:return self.config["white_list"]except Exception as e:print(f'white list section not exists: {e}')project_config = ProjectConfig()# 后台服务
server_ip, server_port = project_config.get_server()# 是否开启debug
debug = project_config.get_debug_mode()# 启动进程数
workers = project_config.get_workers()# 日志
log_level, log_dir, log_max_age = project_config.get_log()
log_dir = os.path.join(BASE_DIR, log_dir)
if not os.path.exists(log_dir):os.makedirs(log_dir)# mysql数据库
mysql_url = project_config.get_mysql()# 识别服务
identify_ip, identify_port, identify_api_path = project_config.get_identify()# 访问白名单列表
white_list = project_config.get_white_list()# 访问静态文件系统目录
static_dir = os.path.join(BASE_DIR, 'myfiles')
if not os.path.exists(static_dir):os.makedirs(static_dir)

common/logging.py

import os
import logging
from logging.handlers import RotatingFileHandlerfrom uvicorn.config import LOG_LEVELSfrom settings import settings# 创建一个RotatingFileHandler，最多备份5个日志文件，每个日志文件最大5M
file_handler = RotatingFileHandler(os.path.join(settings.log_dir, "uvicorn.log"), encoding='UTF-8', maxBytes=5*1024*1024, backupCount=5)
file_handler.setLevel(LOG_LEVELS[settings.log_level])
file_handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(name)s  %(message)s'))logging.basicConfig(handlers=[file_handler])# 获取Uvicorn的logger并添加文件处理器
logger = logging.getLogger("uvicorn")
logger.setLevel(LOG_LEVELS[settings.log_level])
# logger.addHandler(file_handler)  # 没打印日志到文件中，还没找到原因

uvicorn_config.json

{"version": 1,"disable_existing_loggers": false,"formatters": {"default": {"()": "uvicorn.logging.DefaultFormatter","fmt": "%(asctime)s - %(levelprefix)s %(message)s","use_colors": null},"access": {"()": "uvicorn.logging.AccessFormatter","fmt": "%(asctime)s - %(levelprefix)s %(client_addr)s - \"%(request_line)s\" %(status_code)s"}},"handlers": {"default": {"formatter": "default","class": "logging.StreamHandler","stream": "ext://sys.stderr"},"access": {"formatter": "access","class": "logging.StreamHandler","stream": "ext://sys.stdout"}},"loggers": {"fastapi": {"handlers": ["default"],"level": "DEBUG"},"uvicorn": {"handlers": ["default"],"level": "DEBUG"},"uvicorn.error": {"level": "INFO"},"uvicorn.access": {"handlers": ["access"],"level": "INFO","propagate": false}}
}