实现基本功能

前言

本章整理 Flask 项目的常用基础能力：应用配置、JSON 日志、Prometheus 监控，以及数据库的入口。

公共功能模块建议放在 pkg 目录下，比如配置模块放在 pkg/config，日志模块放在 pkg/log，指标模块放在 pkg/metrics。

应用配置

使用 app.config

Flask 的配置保存在 app.config 中，所有配置项名称都要大写：

app = Flask(__name__)
app.config["MYSQL_PASSWORD"] = "123456"
app.config["TESTING"] = True

print(app.config["MYSQL_PASSWORD"])

使用 Python 配置文件

config.py：

MYSQL_PASSWORD = "123456"

app.py：

import config

app = Flask(__name__)
app.config.from_object(config)

使用 JSON / TOML 文件

Flask 2.0 新增的 from_file：

import json
import tomllib

app.config.from_file("config.json", load=json.load)
app.config.from_file("config.toml", load=tomllib.load)

备注

Python 3.11+ 使用标准库 tomllib 解析 TOML；旧写法 toml.load 来自第三方 toml 包，已不推荐。

多格式配置文件模块

配置参数较多时，可以封装一个按文件后缀自动选择解析器的模块。pkg/config/config.py：

import configparser
import json
import os
import tomllib
from abc import ABCMeta, abstractmethod
from pathlib import Path
from typing import Any

import yaml


class Configer(metaclass=ABCMeta):
    def __init__(self):
        self.config: dict = {}

    @abstractmethod
    def read_config(self):
        pass

    def is_exists(self, config_file: str) -> bool:
        return os.path.exists(config_file)


class JsonConfiger(Configer):
    def __init__(self, config_file: str):
        if not self.is_exists(config_file):
            raise FileNotFoundError(f"{config_file} is not exists")
        self.config_file = config_file
        self.config = self.read_config()

    def read_config(self) -> dict:
        with open(self.config_file, "r", encoding="utf-8") as f:
            return json.load(f)


class IniConfiger(Configer):
    def __init__(self, config_file: str):
        if not os.path.exists(config_file):
            raise FileNotFoundError(f"Config file {config_file} not exists")
        self.config_file = config_file
        self.config = self.read_config()

    def read_config(self) -> dict[str, Any]:
        config = configparser.ConfigParser()
        config.read(self.config_file, encoding="utf-8")
        return {section: dict(config.items(section)) for section in config.sections()}


class YamlConfiger(Configer):
    def __init__(self, config_file: str):
        if not os.path.exists(config_file):
            raise FileNotFoundError(f"Config file {config_file} not found")
        self.config_file = config_file
        self.config = self.read_config()

    def read_config(self) -> dict:
        with open(self.config_file, "r", encoding="utf-8") as f:
            return yaml.safe_load(f.read())


class TomlConfiger(Configer):
    def __init__(self, config_file: str):
        if not os.path.exists(config_file):
            raise FileNotFoundError(f"Config file {config_file} does not exist")
        self.config_file = config_file
        self.config = self.read_config()

    def read_config(self) -> dict:
        with open(self.config_file, "rb") as f:
            return tomllib.load(f)


def new_configer(config_file: str = "") -> Configer:
    p = Path(config_file)
    suffix = p.suffix.lower()
    if suffix in (".toml",):
        return TomlConfiger(config_file)
    if suffix in (".json",):
        return JsonConfiger(config_file)
    if suffix in (".yaml", ".yml"):
        return YamlConfiger(config_file)
    if suffix in (".ini",):
        return IniConfiger(config_file)
    raise ValueError(f"Unsupported config file type: {p.suffix}")

pkg/config/__init__.py 中实例化单例，其它模块直接引用：

from .config import new_configer

config_file = "conf/app.yaml"
configer = new_configer(config_file)

依赖安装：python -m pip install pyyaml。

JSON 日志

Flask 默认在控制台输出非结构化请求日志。要输出 JSON 格式日志并写入单独文件，可以禁用默认请求日志，再通过钩子函数自行记录。

定义日志器

pkg/log/log.py：

import json
import logging
import sys
from logging.handlers import TimedRotatingFileHandler
from pathlib import Path


class JsonFormatter(logging.Formatter):
    """普通业务日志，记录调用位置信息"""

    def format(self, record: logging.LogRecord) -> str:
        log_record = {
            "@timestamp": self.formatTime(record, "%Y-%m-%dT%H:%M:%S%z"),
            "level": record.levelname,
            "name": record.name,
            "file": record.filename,
            "lineno": record.lineno,
            "func": record.funcName,
            "message": record.getMessage(),
        }
        return json.dumps(log_record, ensure_ascii=False)


class AccessLogFormatter(logging.Formatter):
    """访问日志，记录请求路径、状态码、耗时等"""

    def format(self, record: logging.LogRecord) -> str:
        log_record = {
            "@timestamp": self.formatTime(record, "%Y-%m-%dT%H:%M:%S%z"),
            "remote_addr": getattr(record, "remote_addr", ""),
            "scheme": getattr(record, "scheme", ""),
            "method": getattr(record, "method", ""),
            "host": getattr(record, "host", ""),
            "path": getattr(record, "path", ""),
            "status": getattr(record, "status", ""),
            "response_length": getattr(record, "response_length", ""),
            "response_time": getattr(record, "response_time", 0),
        }
        return json.dumps(log_record, ensure_ascii=False)


class FlaskLogger(logging.Logger):
    """自定义日志类：按天轮转、JSON 格式、可选控制台输出"""

    def __init__(
        self,
        name: str = __name__,
        level: int = logging.DEBUG,
        logfile: str = "app.log",
        logdir: str = "",
        access_log: bool = False,
        console: bool = True,
        json_log: bool = True,
    ):
        super().__init__(name, level)
        self.logfile = logfile
        self.logdir = logdir
        self.access_log = access_log
        self.console = console
        self.json_log = json_log
        self.setup_logpath()
        self.setup_handler()

    def setup_logpath(self):
        if not self.logdir:
            return
        p = Path(self.logdir)
        if not p.exists():
            p.mkdir(parents=True, exist_ok=True)
        self.logfile = str(p / self.logfile)

    def setup_handler(self):
        formatter = (
            self.set_json_formatter() if self.json_log else self.set_plain_formatter()
        )
        self.addHandler(self.set_handler_file(formatter))
        if self.console:
            self.addHandler(self.set_handler_stdout(formatter))

    def set_plain_formatter(self):
        fmt = "%(asctime)s | %(levelname)s | %(name)s | %(filename)s:%(lineno)d | %(funcName)s | %(message)s"
        return logging.Formatter(fmt, datefmt="%Y-%m-%dT%H:%M:%S%z")

    def set_json_formatter(self):
        return AccessLogFormatter() if self.access_log else JsonFormatter()

    def set_handler_stdout(self, formatter: logging.Formatter):
        handler = logging.StreamHandler(sys.stdout)
        handler.setFormatter(formatter)
        return handler

    def set_handler_file(self, formatter: logging.Formatter):
        handler = TimedRotatingFileHandler(
            filename=self.logfile,
            when="midnight",
            interval=1,
            backupCount=7,
            encoding="utf-8",
        )
        handler.setFormatter(formatter)
        return handler

pkg/log/__init__.py：

from .log import FlaskLogger

access_logger = FlaskLogger("access", logdir="logs", access_log=True, logfile="access.log")
logger = FlaskLogger(logdir="logs")

__all__ = ["access_logger", "logger"]

钩子函数记录请求日志

import time

from flask import g, request, Response


@app.before_request
def start_timer():
    g.start_time = time.time()


@app.after_request
def log_request(response: Response):
    response_length = (
        response.content_length if response.content_length is not None else "-"
    )
    log_message = {
        "remote_addr": request.remote_addr,
        "method": request.method,
        "scheme": request.scheme,
        "host": request.host,
        "path": request.path,
        "status": response.status_code,
        "response_length": response_length,
        "response_time": round(time.time() - g.start_time, 4),
    }
    access_logger.info("", extra=log_message)
    return response

访问日志中间件（WSGI 层）

如果希望对 /metrics、/health 等路径跳过访问日志，可以使用 WSGI 中间件实现：

import time

from pkg.log import access_logger


class AccessLogMiddleware:
    def __init__(self, app):
        self.app = app
        self.white_list = frozenset(["/metrics", "/health"])

    def __call__(self, environ, start_response):
        log_entry = {
            "remote_addr": environ.get("REMOTE_ADDR", "NaN"),
            "method": environ.get("REQUEST_METHOD", "NaN"),
            "scheme": environ.get("wsgi.url_scheme", "NaN"),
            "host": environ.get("HTTP_HOST", "NaN"),
            "path": environ.get("PATH_INFO", "NaN"),
            "status": None,
            "response_length": None,
            "response_time": None,
        }
        resp_status_code = None
        resp_length = None

        def catching_start_response(status, headers, exc_info=None):
            nonlocal resp_status_code, resp_length
            resp_status_code = status.split(" ")[0]
            for key, value in headers:
                if key == "Content-Length":
                    resp_length = int(value)
                    break
            return start_response(status, headers, exc_info)

        start_time = time.time()
        response = self.app(environ, catching_start_response)
        log_entry["status"] = int(resp_status_code)
        log_entry["response_length"] = resp_length
        log_entry["response_time"] = round(time.time() - start_time, 4)

        if log_entry["path"] not in self.white_list:
            access_logger.info("", extra=log_entry)
        return response

注册：

from pkg.middlewares import AccessLogMiddleware

app = Flask(__name__)
app.wsgi_app = AccessLogMiddleware(app.wsgi_app)

启动时禁用 werkzeug 默认请求日志：

logging.getLogger("werkzeug").disabled = True

Prometheus 监控

服务监控分为日志监控（ELK 等收集）和指标监控（Prometheus 收集响应码、响应时间等可量化指标）。

安装 SDK：

python -m pip install prometheus-client

方式一：make_wsgi_app 挂载 /metrics

from flask import Flask
from prometheus_client import Info, make_wsgi_app
from werkzeug.middleware.dispatcher import DispatcherMiddleware

app = Flask(__name__)

i = Info("my_build_version", "Description of info")
i.info({"version": "1.2.3", "buildhost": "foo@bar"})

app.wsgi_app = DispatcherMiddleware(app.wsgi_app, {"/metrics": make_wsgi_app()})


@app.get("/")
def hello():
    return "Hello World!"

访问 http://127.0.0.1:5000/metrics 即可看到指标输出。默认会输出 Python GC、进程等信息；如果不想要这些输出，可以自定义 CollectorRegistry：

from prometheus_client import (
    GC_COLLECTOR,
    PLATFORM_COLLECTOR,
    PROCESS_COLLECTOR,
    CollectorRegistry,
    Info,
    make_wsgi_app,
)

registry = CollectorRegistry(auto_describe=True)
registry.register(PROCESS_COLLECTOR)
registry.register(GC_COLLECTOR)
registry.register(PLATFORM_COLLECTOR)

i = Info("my_build_version", "Description of info", registry=registry)
i.info({"version": "1.2.3", "buildhost": "foo@bar"})

app.wsgi_app = DispatcherMiddleware(
    app.wsgi_app, {"/metrics": make_wsgi_app(registry=registry)}
)

方式二：中间件自动采集请求指标

import socket
import sys
import time
from functools import lru_cache

from prometheus_client import Counter, Histogram, Info


class Metrics:
    app_info = Info("app_info", "Application info")
    request_counter = Counter(
        "http_requests_total", "Total HTTP Requests", ["method", "path", "status"]
    )
    response_time = Histogram(
        "http_response_time_seconds", "HTTP Response Time", ["method", "path"]
    )

    def __init__(self):
        self.app_info.info(
            {
                "version": self._get_version(),
                "host": self.localip(),
                "python_version": self.get_pyversion(),
            }
        )

    def _get_version(self) -> str:
        return "0.1.0"

    @lru_cache(maxsize=128)
    def localip(self):
        try:
            s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            s.connect(("8.8.8.8", 53))
            return s.getsockname()[0]
        except Exception:
            return "127.0.0.1"

    @lru_cache(maxsize=128)
    def get_pyversion(self):
        return sys.version.split(" ")[0]


collector = Metrics()


class MetricsMiddleware:
    def __init__(self, app):
        self.app = app
        self.white_list = frozenset(["/metrics", "/health"])

    def __call__(self, environ, start_response):
        method = environ.get("REQUEST_METHOD", "NaN")
        path = environ.get("PATH_INFO", "NaN")
        resp_status_code = None

        def catching_start_response(status, headers, exc_info=None):
            nonlocal resp_status_code
            resp_status_code = status.split(" ")[0]
            return start_response(status, headers, exc_info)

        start_time = time.time()
        response = self.app(environ, catching_start_response)
        response_time = round(time.time() - start_time, 4)

        if path not in self.white_list:
            collector.request_counter.labels(method, path, resp_status_code).inc()
            collector.response_time.labels(method, path).observe(response_time)
        return response


app.wsgi_app = DispatcherMiddleware(
    app.wsgi_app, {"/metrics": make_wsgi_app()}
)
app.wsgi_app = MetricsMiddleware(app.wsgi_app)

连接数据库

• 关系型数据库（Flask-SQLAlchemy）：见 连接数据库。
• Redis：可使用 redis-py + Flask-Caching（见 补充：使用Flask-Caching缓存数据）。