[Python] Flask生产级项目模板实战：统一返回、日志、异常与JSON编解码最佳实践

本文将介绍一个为实际项目需求而提炼出的 Python Flask 服务模板。该模板旨在为基于Flask开发的Web API服务提供一套开箱即用、易于扩展的基础能力，帮助开发者快速搭建规范、健壮的后端服务。

模板的核心特性包括：

• 统一响应格式：规范API返回数据结构。
• 自定义JSON编解码：优雅处理Python复杂对象（如datetime、numpy数组等）的序列化。
• 生产级日志：支持控制台与文件双输出，并具备日志文件自动轮转与清理能力。
• 全局异常捕获：统一处理未捕获异常，返回友好错误信息。
• 请求/响应日志：自动记录详细的请求与响应信息，便于调试与审计。

项目代码已开源在GitHub，后续通用的增强功能也会持续更新在该仓库中。

统一响应数据格式

规范化的API响应有助于前后端协作。模板定义了 Result 类，统一返回包含 success、result、message 三个字段的JSON数据。

代码位置：src/controller/result.py

class Result(dict):
    """
    结果对象
    """
    def __init__(self, isSuccess: bool, result: any = None, message: str = None):
        super().__init__()
        self["success"] = "success" if isSuccess else "fail"
        self["result"] = result
        self["message"] = message

    @classmethod
    def success(cls, result: any = None):
        return cls(isSuccess=True, result=result)

    @classmethod
    def fail(cls, message: str = None):
        return cls(isSuccess=False, message=message)

使用方式：

• 成功返回：return Result.success(data)
• 失败返回：return Result.fail(“错误信息”)

自定义JSON编解码器

直接序列化Python中的datetime、Enum或numpy对象时会报错。本模板提供了统一的编解码方案，支持Flask内置序列化和标准json模块。

代码位置：src/utils/json_codec.py
核心是一个codec函数，可根据需要扩展支持的对象类型。

def codec(obj):
    """ 自定义编解码器，特殊对象转换 """
    if isinstance(obj, datetime):
        # 处理日期时间对象
        return obj.strftime("%Y-%m-%d %H:%M:%S")
    elif isinstance(obj, Enum):
        # 处理枚举对象
        return obj.name
    elif isinstance(obj, np.ndarray):
        # 处理numpy数组
        return obj.tolist()
    elif isinstance(obj, np.integer):
        # 处理numpy整数类型
        return int(obj)
    elif isinstance(obj, np.floating):
        # 处理numpy浮点数类型
        return float(obj)
    return None

class CustomJSONProvider(DefaultJSONProvider):
    """ 自定义编解码器, 用于flask """
    def default(self, obj):
        codec_result = codec(obj)
        return codec_result if codec_result else super(CustomJSONProvider, self).default(obj)

class CustomEncoder(json.JSONEncoder):
    """ 自定义编解码器，用于 json 模块 """
    def default(self, obj):
        codec_result = codec(obj)
        return codec_result if codec_result else super(CustomEncoder, self).default(obj)

使用方式：

• 在Flask App中：app.json = CustomJSONProvider(app)
• 使用json模块时：json.dumps(data, cls=CustomEncoder)

生产级日志配置

该配置将sys.stdout和sys.stderr重定向至自定义的日志处理器，实现以下功能：

1. 日志同时输出到控制台和指定文件。
2. 日志格式包含时间戳和线程信息。
3. 自动检查日志文件大小，超过100MB时自动截断，保留最新部分。

代码位置：src/config/log_config.py

class LogConfig:
    def __init__(self, name: any, log_type: str, log_file: str):
        self.log_file = log_file
        self.log_type = log_type
        self.max_size = 100 * 1024 * 1024  # 100MB
        self.trim_size = 99 * 1024 * 1024  # 保留99MB
        self._last_check = time.time()
        # 创建专用的日志记录器
        self.logger = logging.getLogger(name)
        self.logger.setLevel(logging.INFO)
        if not self.logger.handlers:
            handler = logging.FileHandler(log_file)
            formatter = logging.Formatter('%(asctime)s - %(message)s')
            handler.setFormatter(formatter)
            self.logger.addHandler(handler)

    def check_and_trim_file(self):
        """检查文件大小并在必要时截断"""
        if os.path.exists(self.log_file):
            file_size = os.path.getsize(self.log_file)
            if file_size > self.max_size:
                try:
                    with open(self.log_file, 'rb') as f:
                        f.seek(self.trim_size)
                        remaining_data = f.read()
                    with open(self.log_file, 'wb') as f:
                        f.write(remaining_data)
                except Exception as e:
                    pass # 截断失败不影响主流程

    def write(self, message):
        if message.strip():
            # 每30秒检查一次文件大小
            current_time = time.time()
            if (current_time - self._last_check) > 30:
                self.check_and_trim_file()
                self._last_check = current_time

            thread_name = threading.current_thread().name
            formatted_message = f”{thread_name} - {message.strip()}”
            self.logger.info(formatted_message) # 写入文件

            # 同时输出到控制台（原始stdout/stderr）
            timestamp = time.time()
            formatted_time = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(timestamp))
            if ‘err’ in self.log_type:
                sys.__stderr__.write(f'{formatted_time} - {formatted_message}\n')
            else:
                sys.__stdout__.write(f'{formatted_time} - {formatted_message}\n')

    def flush(self):
        if ‘err’ in self.log_type:
            sys.__stderr__.flush()
        else:
            sys.__stdout__.flush()

def log_config(name: any, log_filename: str):
    """配置日志，重定向系统输出"""
    try:
        sys.stdout = LogConfig(name=name, log_type='out', log_file=log_filename)
        sys.stderr = LogConfig(name=name, log_type='err’, log_file=log_filename)
    except Exception as e:
        traceback.print_exc()
        print(f“日志重定向配置失败: {e}”)

全局异常捕获与请求日志

通过继承Flask类并添加钩子函数，实现全局异常处理和详细的请求/响应日志记录。

代码位置：src/application.py (部分关键方法)

def handle_exception(self, ex: Exception):
    """全局异常处理器，捕获所有未处理的异常"""
    traceback.print_exc()  # 打印堆栈到日志
    return Result.fail(f'{ex}'), 500 # 返回统一格式的错误信息

def log_request_info(self):
    """在请求开始时记录详细信息"""
    self.request_duration_local.start_time = datetime.now()
    # 记录请求路径、方法、头部、参数等
    # ... (具体日志格式代码，同上文)
    print(log_message)

def process_response(self, response):
    """在请求结束后记录响应信息和耗时"""
    duration = datetime.now() - self.request_duration_local.start_time
    # 记录响应状态码、耗时、响应体（过长会截断）
    # ... (具体日志格式代码，同上文)
    print(log_message)
    return response

定制化Flask应用对象

为了方便集成所有功能，模板提供了一个FlaskApp类，它继承自原生Flask，并在初始化时完成所有配置。

代码位置：src/application.py

class FlaskApp(Flask):
    def __init__(self, import_name: str, log_filename: str, **kwargs):
        super().__init__(import_name, **kwargs)
        # 1. 配置日志
        log_config(name=import_name, log_filename=log_filename)
        # 2. 初始化蓝图注册列表
        self.blueprints_to_register = []
        from src.controller.test_controller import test_bp
        self.add_blueprint(bp=test_bp)
        # 3. 注册请求钩子
        self.before_request(self.log_request_info)
        self.after_request(self.process_response)
        # 4. 注册全局异常处理器
        self.errorhandler(Exception)(self.handle_exception)
        # 5. 设置自定义JSON编码器
        self.json = CustomJSONProvider(self)
        # 6. 用于记录请求耗时的线程局部变量
        self.request_duration_local = threading.local()

    def add_blueprint(self, bp: Blueprint):
        """添加需要注册的蓝图"""
        self.blueprints_to_register.append(bp)

    def set_cors_and_register_blueprint(self, bp: Blueprint, cors_config: dict = None):
        """为蓝图配置CORS并注册"""
        if cors_config is None:
            cors_config = {“origins”: “*”, “allow_headers”: [“Content-Type”], “methods”: [“GET”, “POST”, “PUT”, “DELETE”, “OPTIONS”, “PATCH”]}
        CORS(bp, **cors_config)
        self.register_blueprint(bp)

    def run(self, host: str | None = None, port: int | None = None,
            debug: bool | None = None, load_dotenv: bool = True, **options) -> None:
        """重写run方法，在启动前统一注册所有已添加的蓝图"""
        for bp in self.blueprints_to_register:
            self.set_cors_and_register_blueprint(bp)
        super().run(host=host, port=port, debug=debug, load_dotenv=load_dotenv, **options)

应用入口示例：

if __name__ == "__main__":
    log_filename = './python.log'
    app = FlaskApp(__name__, log_filename)
    # 可根据环境判断是否开启debug模式，此处示例为macOS开发环境开启
    app.run(host=“0.0.0.0”, port=5001, debug=platform.system() == 'Darwin')

测试接口示例

以下是一个测试蓝图示例，演示了如何定义接口，并测试统一响应、JSON编码、日志和异常捕获功能。

代码位置：src/controller/test_controller.py

from flask import Blueprint, request
from datetime import datetime
import numpy as np
from .result import Result

test_bp = Blueprint(“test”, __name__, url_prefix=“/test”)

@test_bp.post(“post”)
def test_post():
    """测试POST请求及复杂对象序列化"""
    request_body = request.json
    data = {
        “request_body”: request_body,
        “request_time”: datetime.now(),
        “ndarray”: np.array([1, 2, 3]),
    }
    print(data)
    return Result.success(data)

@test_bp.get(“current_time”)
def current_time():
    """测试GET请求及日志"""
    args = request.args
    data = {
        “args”: args,
        “request_time”: datetime.now(),
    }
    print(data)
    return Result.success(data)

@test_bp.get(“error”)
def test_error():
    """测试全局异常捕获"""
    raise Exception(“这是一个测试异常”)

使用建议与开源地址

此模板提供的是一套可复用的模式而非固定框架，建议开发者根据自身项目需求，有选择地复制相关模块代码集成到自己的项目中，而不是直接克隆整个仓库作为项目基础。

项目已开源，旨在持续积累Flask服务开发的通用解决方案。访问 Python 板块可以了解更多后端开发相关内容。