简单描述

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

main.py

import signal
import sys
from contextlib import asynccontextmanager

from fastapi import FastAPI
import uvicorn
from fastapi.staticfiles import StaticFiles

from settings import settings
from routers.xxx import xxx_router
from common.logging import logger


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Code to run on startup
    logger.info("Starting up...")
    # You can initialize resources here (e.g., database connections)

    yield

    # Code to run on shutdown
    logger.info("Shutting down...")
    # Clean up resources here

app = 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):
    # 如果程序是被打包成了单一文件，这个条件是True
    BASE_DIR = os.path.dirname(sys.executable)
else:
    # 如果程序是直接运行的.py文件，这个条件是True
    BASE_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()

    @staticmethod
    def get_config_path():
        config_path = os.path.join(BASE_DIR, 'config.json')
        if os.path.exists(config_path):
            return config_path
        print(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 False
        except 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 RotatingFileHandler

from uvicorn.config import LOG_LEVELS

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