Python 基础体系 · 第 78/112 篇。示例统一以 Python 3.14 为语言基线；第三方库使用与其兼容的现代稳定版本，版本敏感行为会单独说明。

Flask 完整基础：应用工厂、Context、Blueprint、扩展和部署

Flask 是一个基于 WSGI 的 Python Web 应用框架。它负责把 HTTP 请求交给路由和视图函数处理，再把视图函数的返回值转换成 HTTP 响应；真正监听端口、接收网络连接的，通常是开发阶段的 Werkzeug 服务器或生产环境中的 Gunicorn、Waitress 等 WSGI 服务器。(flask.palletsprojects.com)

本文使用 Python 3.14 语法和工程环境说明 Flask 的核心基础，重点解决五个问题：

如何用应用工厂创建可测试、可复用的 Flask 应用；
Context 为什么存在，以及 current_app、request、session、g 到底指向什么；
Blueprint 如何拆分路由，又为什么它不是一个独立应用；
扩展如何与应用工厂配合，避免循环导入和全局状态污染；
Flask 应用如何从本地开发运行到生产部署，并理解 WSGI 与 ASGI 的边界。

一、先建立 Flask 的整体模型

一个最小 Flask 应用可以写成：

from flask import Flask

app = Flask(__name__)


@app.get("/")
def index():
    return {"message": "hello, flask"}

这里有三个对象需要区分：

Flask 实例：应用对象，也是 WSGI 应用；
路由规则：例如 GET /；
视图函数：收到匹配请求后执行的 Python 函数。

Flask 官方示例中，Flask(__name__) 的参数用于帮助 Flask 定位模板、静态文件等资源；@app.route() 或 @app.get() 则把 URL 规则注册到应用对象上。(flask.palletsprojects.com)

从服务器到 Flask 的基本数据流是：

sequenceDiagram
    participant C as HTTP 客户端
    participant S as WSGI 服务器
    participant A as Flask 应用
    participant V as 视图函数

    C->>S: HTTP 请求
    S->>S: 转换为 WSGI environ
    S->>A: app(environ, start_response)
    A->>A: 创建 AppContext 和 RequestContext
    A->>A: URL 匹配
    A->>V: 调用视图函数
    V-->>A: 返回字符串/字典/Response
    A->>A: 生成响应并执行 after/teardown
    A-->>S: status、headers、body
    S-->>C: HTTP 响应

WSGI 应用的核心调用形式可以抽象为：

response_iterable = app(environ, start_response)

其中：

environ 是服务器构造的请求环境字典；
start_response(status, headers) 用于设置状态码和响应头；
返回值是可迭代的响应体。

Flask 本身不应该直接暴露给公网。开发服务器的目标是调试便利性，不是生产环境的安全性、稳定性或效率。生产环境应使用专用 WSGI 服务器，并通常在前面放置反向代理。(flask.palletsprojects.com)

二、应用工厂：把“创建应用”变成函数

2.1 什么是应用工厂

应用工厂是一个负责创建并配置 Flask 实例的函数，典型名称是 create_app：

from flask import Flask


def create_app(test_config=None):
    app = Flask(__name__)

    app.config.from_mapping(
        SECRET_KEY="dev-only",
        JSON_SORT_KEYS=False,
    )

    if test_config is not None:
        app.config.update(test_config)

    @app.get("/")
    def index():
        return {"message": "hello"}

    return app

使用时：

app = create_app()

Flask CLI 可以自动识别名为 create_app 或 make_app 的工厂函数：

flask --app myproject run --debug

如果工厂函数需要参数，也可以显式调用：

flask --app 'myproject:create_app(testing=True)' run

Flask 官方推荐工厂模式的主要原因是：同一个进程中可以创建多个配置不同的应用实例，尤其适合测试、不同环境配置和多实例场景。(flask.palletsprojects.com)

2.2 直接全局创建应用的问题

下面的写法在小程序中没有问题：

app = Flask(__name__)

from .views import user_view
app.register_blueprint(user_view)

但随着项目变大，会出现几个结构性问题。

第一，导入模块时就创建应用，导致测试难以替换配置：

# 错误倾向：导入时绑定固定数据库
db = SQLAlchemy(app)

测试时想换成内存数据库或临时数据库，就必须修改已经创建的全局对象。

第二，模块之间容易形成循环导入：

app.py -> views.py -> models.py -> app.py

第三，导入应用模块本身就会执行所有初始化逻辑。某些初始化逻辑依赖环境变量、数据库或外部服务时，单纯导入模块可能产生副作用。

应用工厂将这些步骤推迟到显式调用时：

调用 create_app()
    ├── 创建 Flask 实例
    ├── 加载配置
    ├── 初始化扩展
    ├── 注册 Blueprint
    ├── 注册错误处理器
    └── 返回应用实例

2.3 一个更完整的配置加载顺序

建议区分三类配置：

代码中的安全默认值；
部署环境提供的配置；
测试显式覆盖的配置。

import os

from flask import Flask


def create_app(test_config=None):
    app = Flask(
        __name__,
        instance_relative_config=True,
    )

    app.config.from_mapping(
        SECRET_KEY=os.environ.get("SECRET_KEY", "dev-only"),
        DATABASE_URL=os.environ.get(
            "DATABASE_URL",
            "sqlite:///default.sqlite3",
        ),
    )

    if test_config is None:
        app.config.from_pyfile("config.py", silent=True)
    else:
        app.config.update(test_config)

    return app

这里的关键是覆盖顺序：

默认配置
    ↓
instance/config.py
    ↓
test_config

测试配置通常应拥有最高优先级，否则测试仍可能连接真实数据库。

不要把生产密钥写成：

SECRET_KEY = "production-secret"

因为源代码、代码仓库备份、构建产物和日志都有泄露风险。开发默认值只用于本地，生产配置应由环境变量、密钥管理系统或部署平台注入。

Flask 的配置对象在应用设置阶段会影响路由、扩展和调试行为；例如 DEBUG 更适合通过 flask --debug 设置，而不是应用已经开始配置后再修改。(flask.palletsprojects.com)

2.4 工厂必须保持“可重复创建”

一个合格的工厂至少应满足：

app1 = create_app({"TESTING": True})
app2 = create_app({"TESTING": False})

assert app1 is not app2
assert app1.testing is True
assert app2.testing is False

这意味着工厂中不应把某个应用实例缓存成隐式全局状态：

_cached_app = None


def create_app():
    global _cached_app

    if _cached_app is None:
        _cached_app = Flask(__name__)

    return _cached_app

这种缓存破坏了测试隔离。第一个测试修改的配置可能影响第二个测试，多个应用实例也无法真正共存。

但是，“工厂可重复创建”不等于所有资源都必须每次重新连接。扩展对象可以复用，数据库连接池也可以由扩展根据应用生命周期管理；真正需要避免的是把某个具体应用的配置和状态错误地绑定到模块级单例中。

三、Context：为什么视图函数可以直接使用 `request`

Flask 中常见的代码是：

from flask import request


@app.get("/search")
def search():
    keyword = request.args.get("q", "")
    return {"keyword": keyword}

request 看起来像一个全局变量，但它不是一个固定的请求对象，而是一个 上下文局部代理。

3.1 Context 的问题背景

假设服务器同时处理两个请求：

请求 A：GET /users/1
请求 B：GET /users/2

如果 Flask 使用普通模块级变量保存当前请求：

current_request = None

处理过程可能变成：

线程 1：current_request = 请求 A
线程 2：current_request = 请求 B
线程 1：读取 current_request，结果变成请求 B

因此，Flask 必须让“当前请求”只对当前执行上下文可见。

现代 Flask 使用 Python 的 contextvars 和 Werkzeug 的 LocalProxy 管理这些上下文局部对象。不同线程或其他执行单元可以通过同一个代理名访问各自的上下文数据。(flask.palletsprojects.com)

3.2 两种 Context

Flask 主要有两类上下文。

Application Context

应用上下文表示当前正在使用哪个 Flask 应用，以及该应用范围内的数据：

current_app：当前 Flask 应用；
g：当前应用上下文期间的临时命名空间。

from flask import current_app, g


def get_feature_flag():
    return current_app.config["FEATURE_ENABLED"]


def cache_user(user):
    g.user = user

g 中的数据只在当前上下文存在。它不是跨请求缓存，也不是进程级全局变量。

Request Context

请求上下文表示当前 HTTP 请求：

request：当前请求对象；
session：当前用户会话。

from flask import request, session


@app.post("/login")
def login():
    username = request.form["username"]
    session["username"] = username
    return {"username": username}

请求上下文通常嵌套在应用上下文中：

AppContext
└── RequestContext
    ├── request
    └── session

处理请求时，Flask 会自动推入请求上下文；请求上下文又会在需要时推入应用上下文。请求结束时，先弹出请求上下文，再弹出应用上下文。(flask.palletsprojects.com)

3.3 为什么应用上下文和请求上下文要分开

应用上下文不一定需要 HTTP 请求。例如 CLI 命令、数据库迁移或后台初始化任务可能只需要知道当前应用：

with app.app_context():
    print(current_app.config["DATABASE_URL"])

请求上下文则必须包含具体请求：

with app.test_request_context(
    "/search?q=flask",
    method="GET",
):
    print(request.path)
    print(request.args["q"])

这两个上下文的含义不同：

对象	需要的上下文	表示
`current_app`	应用上下文	当前 Flask 应用
`g`	应用上下文	当前上下文的临时数据
`request`	请求上下文	当前 HTTP 请求
`session`	请求上下文	当前请求对应的会话

3.4 上下文外访问为什么报错

下面的代码会失败：

from flask import current_app

print(current_app.config)

典型错误是：

RuntimeError: Working outside of application context.

原因不是 current_app 没有定义，而是它无法找到“当前绑定的 Flask 应用”。

正确方式是手动推入上下文：

app = create_app()

with app.app_context():
    print(current_app.config["SECRET_KEY"])

数据库初始化、命令行任务和测试中的数据库操作经常需要这种写法。Flask 官方也建议在配置应用或初始化资源时使用 app.app_context()。(flask.palletsprojects.com)

3.5 `g` 的正确用途：一次请求内复用资源

下面实现一个简单的数据库连接缓存：

import sqlite3

from flask import current_app, g


def get_db():
    if "db" not in g:
        g.db = sqlite3.connect(
            current_app.config["DATABASE_PATH"]
        )
        g.db.row_factory = sqlite3.Row

    return g.db


def close_db(exception=None):
    db = g.pop("db", None)
    if db is not None:
        db.close()

在工厂中注册清理函数：

def create_app(test_config=None):
    app = Flask(__name__)

    app.config.from_mapping(
        DATABASE_PATH="app.sqlite3",
    )

    if test_config:
        app.config.update(test_config)

    app.teardown_appcontext(close_db)

    return app

一次请求内：

@app.get("/users/<int:user_id>")
def get_user(user_id):
    db1 = get_db()
    db2 = get_db()

    assert db1 is db2

    row = db1.execute(
        "SELECT id, name FROM users WHERE id = ?",
        (user_id,),
    ).fetchone()

    if row is None:
        return {"error": "user not found"}, 404

    return dict(row)

第一次调用 get_db() 时连接被放入 g；第二次调用直接复用；请求结束时 teardown_appcontext 取出连接并关闭。

这里存在一个重要边界：

g.db

表示的是“当前应用上下文中的数据库连接”，不是：

全局共享数据库连接

如果把连接放到模块级变量中，多个线程可能同时使用同一连接，连接的线程安全、事务边界和异常恢复都会变得复杂。

3.6 生命周期与异常

一个简化的请求生命周期是：

推入 AppContext
推入 RequestContext
    before_request
    URL 匹配
    视图函数
    errorhandler（如果发生异常）
    转换为 Response
    after_request
    保存 session
弹出 RequestContext
    teardown_request
弹出 AppContext
    teardown_appcontext

teardown_request 和 teardown_appcontext 的职责是释放资源，而不是判断请求是否成功。它们即使在未处理异常后也会运行，因此清理逻辑必须能够处理异常路径。(flask.palletsprojects.com)

例如：

@app.teardown_appcontext
def close_resources(exception=None):
    resource = g.pop("resource", None)
    if resource is not None:
        resource.close()

不要在 teardown 中假设业务一定成功：

@app.teardown_appcontext
def bad_cleanup(exception=None):
    g.db.commit()

如果业务异常，teardown 阶段再提交事务可能造成不可预期的状态。提交或回滚应由明确的事务边界控制，teardown 主要负责释放连接、游标、文件句柄等资源。

四、Blueprint：应用内部的模块化注册机制

4.1 Blueprint 是什么

Blueprint 可以理解为一组“等待注册到应用上的操作”：

from flask import Blueprint

users_bp = Blueprint(
    "users",
    __name__,
    url_prefix="/users",
)

然后在 Blueprint 上定义路由：

@users_bp.get("/<int:user_id>")
def get_user(user_id):
    return {"user_id": user_id}

最后在工厂中注册：

app.register_blueprint(users_bp)

完整路径是：

Blueprint URL：/<int:user_id>
前缀：/users
最终 URL：/users/<int:user_id>

Blueprint 不是 Flask 应用，也不是独立的 WSGI 应用。它保存的是路由、错误处理器、模板过滤器、静态文件等注册操作，注册后会修改目标 Flask 应用。一个 Blueprint 也可以注册到多个应用，甚至多次注册。(flask.palletsprojects.com)

4.2 推荐的目录结构

一个小型但可扩展的项目可以这样组织：

myproject/
├── pyproject.toml
├── myproject/
│   ├── __init__.py
│   ├── extensions.py
│   ├── users/
│   │   ├── __init__.py
│   │   └── routes.py
│   └── health/
│       ├── __init__.py
│       └── routes.py
└── tests/
    ├── conftest.py
    └── test_health.py

health/routes.py：

from flask import Blueprint

bp = Blueprint("health", __name__)


@bp.get("/healthz")
def healthz():
    return {
        "status": "ok",
    }

health/__init__.py：

from .routes import bp

__all__ = ["bp"]

myproject/__init__.py：

from flask import Flask

from .health import bp as health_bp


def create_app(test_config=None):
    app = Flask(__name__)

    app.config.from_mapping(
        SECRET_KEY="dev-only",
    )

    if test_config is not None:
        app.config.update(test_config)

    app.register_blueprint(health_bp)

    return app

启动：

flask --app myproject run --debug

请求：

curl http://127.0.0.1:5000/healthz

预期响应：

{"status":"ok"}

4.3 为什么 Blueprint 中使用 `current_app`

Blueprint 被导入时，应用可能还不存在：

# routes.py 被 import 时，create_app() 可能尚未调用

因此不要在模块导入阶段读取应用配置：

# 不推荐
TIMEOUT = app.config["TIMEOUT"]

应该在请求处理或注册阶段读取：

from flask import Blueprint, current_app

bp = Blueprint("users", __name__)


@bp.get("/config")
def show_config():
    return {
        "timeout": current_app.config["TIMEOUT"],
    }

此时请求已经推入应用上下文，current_app 才能定位当前应用。

4.4 Blueprint 级别的钩子和错误处理

Blueprint 可以拥有自己的请求钩子：

from flask import Blueprint, g, request

bp = Blueprint("admin", __name__, url_prefix="/admin")


@bp.before_request
def require_admin():
    token = request.headers.get("X-Admin-Token")

    if token != "local-admin-token":
        return {"error": "admin required"}, 403

    g.is_admin = True

这个钩子只针对属于该 Blueprint 的路由。它返回响应后，后续视图函数不会执行。

Blueprint 也可以注册错误处理器：

@bp.errorhandler(404)
def users_not_found(error):
    return {
        "error": "users resource not found",
    }, 404

但需要注意：路由匹配发生在 Blueprint 逻辑之前，因此某些全局 404 无法被某个 Blueprint 单独捕获。也就是说，请求根本没有匹配到该 Blueprint 的路由时，Flask 未必知道应该使用哪个 Blueprint 的 404 处理器。

4.5 Blueprint 与 URL、端点名称

Blueprint 名称会参与端点名生成：

bp = Blueprint("users", __name__)


@bp.get("/<int:user_id>")
def detail(user_id):
    ...

端点名通常是：

users.detail

因此可以这样生成 URL：

from flask import url_for

url_for("users.detail", user_id=42)

如果两个 Blueprint 中都有名为 detail 的函数，由于端点名包含 Blueprint 名称，它们不会直接冲突：

users.detail
orders.detail

但 Blueprint 名称本身必须在应用中保持唯一，否则注册时会出现端点或名称冲突。

五、扩展：延迟绑定应用，而不是绑定某个全局应用

5.1 扩展的两阶段初始化

Flask 扩展通常遵循两阶段模式：

# extensions.py
from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

然后在工厂中绑定：

# __init__.py
from flask import Flask

from .extensions import db


def create_app(test_config=None):
    app = Flask(__name__)

    app.config.from_mapping(
        SQLALCHEMY_DATABASE_URI="sqlite:///app.sqlite3",
    )

    if test_config:
        app.config.update(test_config)

    db.init_app(app)

    return app

不要这样写：

# 不推荐
db = SQLAlchemy(app)

因为这会在导入阶段把扩展绑定到一个具体应用。使用工厂时，扩展对象应该先独立创建，再通过 init_app(app) 绑定当前应用。这样同一个扩展对象可以服务于多个应用实例。(flask.palletsprojects.com)

可以把扩展看作：

扩展对象：保存扩展代码和共享定义
init_app(app)：把当前应用的配置、钩子和资源接入扩展

这种设计解决的是依赖方向问题：

extensions.py      不依赖具体 app
create_app()       创建 app，然后初始化 extensions
routes.py          通过 current_app 使用当前 app

5.2 扩展初始化顺序

典型顺序如下：

def create_app(test_config=None):
    app = Flask(__name__)

    load_config(app, test_config)

    db.init_app(app)
    cache.init_app(app)
    login_manager.init_app(app)

    register_error_handlers(app)
    register_blueprints(app)
    register_cli_commands(app)

    return app

配置必须先于扩展初始化，因为扩展通常会在 init_app 阶段读取配置：

创建应用
    ↓
加载 DATABASE_URL、SECRET_KEY 等配置
    ↓
db.init_app(app)
    ↓
注册依赖数据库的 Blueprint

如果先执行：

db.init_app(app)
app.config["DATABASE_URL"] = "..."

扩展可能已经读取了旧配置，之后修改不一定生效。

5.3 扩展不能替代业务边界

扩展负责提供能力，例如：

数据库连接和 ORM；
缓存；
登录会话；
表单校验；
任务队列；
API 文档或序列化。

但扩展不应成为所有业务逻辑的容器。例如，下面的视图把认证、查询、业务规则和响应格式全部混在一起：

@bp.post("/orders")
def create_order():
    user = load_user_from_token()
    body = request.get_json()
    validate_order(body)
    order = db.session.execute(...)
    send_email(...)
    return jsonify(order)

更容易测试的结构是：

Blueprint：解析 HTTP 输入、调用服务、生成 HTTP 响应
Service：执行业务规则和事务
Repository：访问数据库
Extension：提供数据库、缓存、队列等基础能力

Flask 不强制这种结构；它提供的是注册机制和请求生命周期，项目可以根据复杂度自行划分模块。Flask 官方也明确表示，框架不会强制项目布局。(github.com)

六、错误处理：把异常转换成稳定的 HTTP 契约

6.1 视图返回值如何变成响应

视图函数可以返回多种形式：

@app.get("/text")
def text():
    return "ok"

@app.get("/json")
def json_response():
    return {"ok": True}

@app.get("/created")
def created():
    return {"id": 1}, 201

from flask import make_response


@app.get("/custom")
def custom():
    response = make_response({"ok": True}, 202)
    response.headers["X-Request-ID"] = "abc"
    return response

Flask 会把这些返回值转换为 Response 对象。视图、before_request 或错误处理器都可能提供响应值；如果 before_request 返回了响应，视图函数就不会执行。(flask.palletsprojects.com)

6.2 为 API 统一错误格式

from flask import Flask
from werkzeug.exceptions import HTTPException


def register_error_handlers(app: Flask):
    @app.errorhandler(HTTPException)
    def handle_http_error(error: HTTPException):
        return {
            "error": {
                "code": error.name.lower().replace(" ", "_"),
                "message": error.description,
            }
        }, error.code

    @app.errorhandler(Exception)
    def handle_unexpected_error(error: Exception):
        app.logger.exception("unexpected error")
        return {
            "error": {
                "code": "internal_server_error",
                "message": "internal server error",
            }
        }, 500

这里要区分两类异常：

HTTPException：例如 404、405、400，通常可以安全转换成对应的 HTTP 响应；
普通 Exception：通常表示程序错误或未预期故障，对客户端只返回通用 500，详细堆栈写入服务端日志。

生产环境不能把异常对象直接返回给客户端：

return {"error": str(error)}, 500

因为异常消息可能包含数据库结构、文件路径、凭据片段或内部服务信息。

6.3 错误处理器与 teardown 的关系

错误处理器负责：

异常 → 对外 HTTP 响应

teardown 负责：

请求结束 → 释放资源

二者不是替代关系：

@app.teardown_appcontext
def cleanup(exception=None):
    close_database_connection()

即使错误处理器把异常变成了 400 或 500，清理函数仍然需要执行。Flask 文档明确说明 teardown 回调独立于请求分发，并且异常路径同样会触发。(flask.palletsprojects.com)

七、完整示例：工厂、Blueprint、扩展和数据库连接

下面给出一个不依赖第三方 ORM 的可运行示例，使用 SQLite 展示核心机制。

7.1 项目结构

demo/
├── pyproject.toml
├── demo/
│   ├── __init__.py
│   ├── db.py
│   └── api.py
└── tests/
    └── test_api.py

7.2 `pyproject.toml`

[project]
name = "flask-context-demo"
version = "0.1.0"
requires-python = ">=3.14"
dependencies = [
    "Flask",
]

[project.optional-dependencies]
test = [
    "pytest",
]

创建虚拟环境并安装：

python3.14 -m venv .venv
source .venv/bin/activate
python -m pip install -e ".[test]"

Windows PowerShell：

py -3.14 -m venv .venv
.venv\Scripts\Activate.ps1
python -m pip install -e ".[test]"

Flask 官方建议为每个项目使用虚拟环境，避免不同项目之间的依赖和 Python 包相互污染。(flask.palletsprojects.com)

7.3 `demo/db.py`

import sqlite3

from flask import current_app, g


def get_db():
    if "db" not in g:
        connection = sqlite3.connect(
            current_app.config["DATABASE"],
        )
        connection.row_factory = sqlite3.Row
        g.db = connection

    return g.db


def close_db(exception=None):
    connection = g.pop("db", None)

    if connection is not None:
        connection.close()


def init_app(app):
    app.teardown_appcontext(close_db)

init_app(app) 是扩展式初始化函数。即使这里只有几行代码，也保留这个结构，未来替换成真正的数据库扩展时，应用工厂不需要改变总体设计。

7.4 `demo/api.py`

from flask import Blueprint, request

from .db import get_db

bp = Blueprint("api", __name__, url_prefix="/api")


@bp.get("/users/<int:user_id>")
def get_user(user_id: int):
    row = get_db().execute(
        "SELECT id, name FROM users WHERE id = ?",
        (user_id,),
    ).fetchone()

    if row is None:
        return {
            "error": {
                "code": "user_not_found",
                "message": "user not found",
            }
        }, 404

    return {
        "id": row["id"],
        "name": row["name"],
    }


@bp.post("/users")
def create_user():
    body = request.get_json(silent=True)

    if not isinstance(body, dict):
        return {
            "error": {
                "code": "invalid_json",
                "message": "request body must be a JSON object",
            }
        }, 400

    name = body.get("name")

    if not isinstance(name, str) or not name.strip():
        return {
            "error": {
                "code": "invalid_name",
                "message": "name must be a non-empty string",
            }
        }, 400

    db = get_db()
    cursor = db.execute(
        "INSERT INTO users (name) VALUES (?)",
        (name.strip(),),
    )
    db.commit()

    return {
        "id": cursor.lastrowid,
        "name": name.strip(),
    }, 201

这里使用 SQL 参数绑定：

"WHERE id = ?", (user_id,)

而不是字符串拼接：

f"WHERE id = {user_id}"

参数绑定让数据库驱动负责处理值的编码和转义，避免 SQL 注入风险。

7.5 `demo/init.py`

import sqlite3
from pathlib import Path

from flask import Flask
from werkzeug.exceptions import HTTPException

from . import db


def create_app(test_config=None):
    app = Flask(__name__)

    app.config.from_mapping(
        SECRET_KEY="development-only",
        DATABASE=str(Path(app.instance_path) / "demo.sqlite3"),
    )

    if test_config is not None:
        app.config.update(test_config)

    Path(app.instance_path).mkdir(
        parents=True,
        exist_ok=True,
    )

    db.init_app(app)

    with app.app_context():
        connection = db.get_db()
        connection.executescript(
            """
            CREATE TABLE IF NOT EXISTS users (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name TEXT NOT NULL
            );
            """
        )
        connection.commit()

    from .api import bp as api_bp

    app.register_blueprint(api_bp)

    @app.errorhandler(HTTPException)
    def handle_http_error(error):
        return {
            "error": {
                "code": error.name.lower().replace(" ", "_"),
                "message": error.description,
            }
        }, error.code

    return app

初始化流程是：

create_app()
    ↓
创建 Flask 实例
    ↓
加载配置
    ↓
创建 instance 目录
    ↓
注册数据库 teardown
    ↓
推入 app context
    ↓
创建 users 表
    ↓
注册 API Blueprint
    ↓
注册错误处理器
    ↓
返回 app

在生产系统中，数据库表结构通常由迁移工具管理，而不是每次启动时执行 CREATE TABLE IF NOT EXISTS。这个示例这样写，是为了让程序在空目录中可以直接运行。

启动：

flask --app demo run --debug

创建用户：

curl \
  -X POST \
  -H 'Content-Type: application/json' \
  -d '{"name":"Alice"}' \
  http://127.0.0.1:5000/api/users

预期响应：

{
  "id": 1,
  "name": "Alice"
}

查询用户：

curl http://127.0.0.1:5000/api/users/1

预期响应：

{
  "id": 1,
  "name": "Alice"
}

查询不存在的用户：

curl http://127.0.0.1:5000/api/users/999

预期响应：

{
  "error": {
    "code": "user_not_found",
    "message": "user not found"
  }
}

八、测试：应用工厂的直接收益

Flask 提供测试客户端，可以在不启动真实 HTTP 服务器的情况下向应用发请求；测试客户端调用的是应用本身，因此适合验证路由、请求解析、响应状态码和上下文行为。(flask.palletsprojects.com)

tests/conftest.py：

import pytest

from demo import create_app


@pytest.fixture()
def app(tmp_path):
    database = tmp_path / "test.sqlite3"

    app = create_app(
        {
            "TESTING": True,
            "DATABASE": str(database),
        }
    )

    yield app


@pytest.fixture()
def client(app):
    return app.test_client()

tests/test_api.py：

def test_create_and_get_user(client):
    response = client.post(
        "/api/users",
        json={"name": "Bob"},
    )

    assert response.status_code == 201
    user_id = response.json["id"]

    response = client.get(f"/api/users/{user_id}")

    assert response.status_code == 200
    assert response.json == {
        "id": user_id,
        "name": "Bob",
    }


def test_invalid_user_name(client):
    response = client.post(
        "/api/users",
        json={"name": ""},
    )

    assert response.status_code == 400
    assert response.json["error"]["code"] == "invalid_name"


def test_missing_user(client):
    response = client.get("/api/users/999")

    assert response.status_code == 404
    assert response.json["error"]["code"] == "user_not_found"

执行：

pytest -q

预期输出类似：

3 passed

如果需要在测试中访问上下文对象：

from flask import g


def test_context(client):
    with client:
        response = client.get("/api/users/999")
        assert response.status_code == 404
        # 此处仍处于测试客户端维护的上下文中

测试客户端可以暂时保留请求上下文，方便检查 session、request 等对象；离开 with 代码块后，上下文会被清理。(flask.palletsprojects.com)

九、WSGI、同步并发与 Flask 的异步视图

9.1 WSGI 的并发模型

Flask 是 WSGI 应用。WSGI 的基本模型是一次调用对应一个请求/响应处理过程：

服务器接收请求
    ↓
调用 Flask WSGI callable
    ↓
Flask 处理请求
    ↓
返回响应可迭代对象

并发能力主要由服务器提供：

多进程：每个进程拥有独立内存；
多线程：同一进程内多个线程处理请求；
gevent 等协作式 worker：通过事件循环或 greenlet 调度；
多实例部署：多个容器或虚拟机共同接收流量。

因此：

@app.get("/slow")
def slow():
    time.sleep(10)
    return {"ok": True}

如果服务器只有一个 worker，这个请求会占用该 worker 十秒。Flask 应用本身不会因为使用了框架就自动获得无限并发。

9.2 Flask 的 `async def` 不等于 ASGI

Flask 支持异步视图：

from flask import Flask

app = Flask(__name__)


@app.get("/async")
async def async_view():
    result = await fetch_remote_data()
    return {"result": result}

但 Flask 仍然是 WSGI 应用。官方文档说明，对于异步视图，Flask 会在线程中启动事件循环执行协程；一次请求仍然占用一个 worker，因此异步视图不会自动提高 Flask 能处理的并发请求数量。异步更适合在同一个请求中并发等待多个 I/O 操作，而不是解决 CPU 密集型计算。(flask.palletsprojects.com)

例如：

@app.get("/aggregate")
async def aggregate():
    user, orders = await asyncio.gather(
        fetch_user(),
        fetch_orders(),
    )

    return {
        "user": user,
        "orders": orders,
    }

这个模式的收益是把两个独立的等待过程重叠起来：

串行：
fetch_user  100 ms
fetch_orders 100 ms
总计约 200 ms

并行等待：
fetch_user  ─┐
              ├── 总计约 100 ms
fetch_orders ─┘

这是在外部服务延迟占主导时的理论收益，实际结果还受连接池、服务器、网络和对方限流影响。

9.3 WSGI 与 ASGI 的差异

WSGI 主要面向传统同步请求/响应；ASGI 将应用调用建模为：

await application(scope, receive, send)

其中：

scope 描述连接或协议上下文；
receive 异步接收事件；
send 异步发送事件。

ASGI 的连接生命周期可以持续多个事件，适合 HTTP、HTTP/2、WebSocket 等协议；ASGI 文档将应用定义为接收和发送异步事件的协程。(asgi.readthedocs.io)

对比：

维度	WSGI	ASGI
应用调用	同步 callable	异步 callable
数据模型	一次请求输入、一次响应输出	scope + receive/send 事件
典型场景	传统 HTTP 请求/响应	异步 HTTP、WebSocket、长连接
Flask 原生定位	WSGI	不是 ASGI 原生框架
并发关键	进程、线程、worker	事件循环、任务、worker

ASGI 的 HTTP 规范设计了 WSGI 兼容路径；同步 WSGI 应用可以通过线程池等方式运行在 ASGI 服务器中，但这并不会让原应用自动变成异步应用。(asgi.readthedocs.io)

9.4 Flask 运行在 ASGI 服务器上

可以使用 asgiref 提供的适配器：

python -m pip install "Flask[async]" asgiref uvicorn

创建 asgi.py：

from asgiref.wsgi import WsgiToAsgi

from demo import create_app

application = WsgiToAsgi(create_app())

启动：

uvicorn asgi:application

这个结构的含义是：

Uvicorn ASGI 服务器
    ↓
WsgiToAsgi 适配器
    ↓
Flask WSGI 应用

它适用于希望把 WSGI 应用接入 ASGI 基础设施的场景，但适配器不能改变 Flask 的 WSGI 生命周期，也不能让同步扩展自动支持异步调用。

如果应用主要依赖：

WebSocket；
长时间保持的连接；
大量并发异步 I/O；
原生 ASGI 中间件；
持续运行的异步后台任务；

则应认真评估 ASGI 原生框架，而不是只把 Flask 套在适配器后面。Flask 官方也指出，异步优先场景可以考虑基于 ASGI 的 Quart。(flask.palletsprojects.com)

十、生产部署：从 Flask CLI 到 WSGI 服务器

10.1 本地开发

开发阶段：

flask --app demo run --debug

开发服务器提供：

代码变更自动重载；
调试器；
便于查看异常堆栈。

但 --debug 会暴露交互式调试器，绝不能在生产环境启用。开发服务器也不是为生产的安全性、稳定性和性能设计的。(flask.palletsprojects.com)

10.2 使用 Waitress

安装：

python -m pip install waitress

应用工厂部署：

waitress-serve --call 'demo:create_app'

如果应用模块中是一个已经创建好的应用对象，则使用：

waitress-serve 'demo:app'

--call 的含义是：导入 demo 模块，找到 create_app，再调用它获得 Flask 应用。Waitress 官方示例使用的格式正是：

{module}:{app}
{module}:{factory}

其中工厂形式需要 --call。(flask.palletsprojects.com)

10.3 使用 Gunicorn

Linux 环境常见命令：

python -m pip install gunicorn
gunicorn 'demo:create_app()'

含义是：

demo       模块
create_app 工厂函数
()         调用工厂

也可以先在模块中暴露一个 WSGI 对象：

# wsgi.py
from demo import create_app

app = create_app()

然后：

gunicorn wsgi:app

Gunicorn 或其他 WSGI 服务器的 worker 数量应根据应用特征、CPU、内存、数据库连接数和实际压测决定，不能套用一个固定公式。

10.4 反向代理的职责

常见生产拓扑：

客户端
  ↓
HTTPS 终止、静态文件、压缩、超时控制
  ↓
Nginx / Apache
  ↓
Gunicorn / Waitress
  ↓
Flask

反向代理可以处理：

TLS 证书；
静态文件；
客户端连接缓冲；
请求体大小限制；
访问日志；
多个后端实例之间的负载均衡。

Flask 官方将“HTTP 服务器位于 WSGI 服务器前方”的结构称为 reverse proxy，并提醒部署在代理后时需要正确告诉 Flask 当前请求经过了代理。(flask.palletsprojects.com)

例如，代理将外部请求：

https://example.com/api/users

转发到内部：

http://127.0.0.1:8000/api/users

如果 Flask 没有正确处理代理转发的协议、主机和前缀信息，可能出现：

生成 http:// 而不是 https:// 的 URL；
重定向地址错误；
客户端 IP 被错误识别；
Cookie 的安全属性判断不正确。

这类配置必须与实际代理拓扑匹配，不能盲目启用 ProxyFix。信任了不受控的 X-Forwarded-* 请求头，会让客户端伪造协议、主机或 IP 信息。

10.5 进程、线程和数据库连接

假设 Gunicorn 启动多个进程：

进程 1：独立内存、独立模块状态
进程 2：独立内存、独立模块状态
进程 3：独立内存、独立模块状态

因此模块级变量：

cache = {}

只在单个进程内共享，不是整个服务集群的共享缓存。

如果需要跨进程共享状态，应使用：

Redis；
数据库；
外部缓存；
消息队列；
其他明确的共享存储。

同时，数据库连接池大小必须与 worker 数量、线程数和数据库最大连接数共同计算。否则应用可能在进程数量增加后，把数据库连接数推过上限，表现为：

连接超时
too many connections
请求延迟突然升高
部分 worker 无法处理新请求

十一、常见误解与诊断路径

11.1 误解：`g` 是全局缓存

错误理解：

g.user = user

等价于：

整个进程永久保存 user

实际情况是：

g 只属于当前 AppContext
请求结束后，g 中的数据就不应继续使用

如果需要跨请求缓存，应使用 Redis、数据库或进程级缓存，并明确失效策略。

11.2 误解：Blueprint 是可独立部署的子应用

Blueprint 没有独立的服务器监听能力，也不能脱离 Flask 应用单独接收请求。它只是应用构建过程的一部分：

Blueprint 定义操作
    ↓
register_blueprint()
    ↓
修改 Flask 应用的路由表和钩子

如果需要多个完全隔离的应用，应创建多个 Flask 实例，再在 WSGI 层使用应用分发，而不是把 Blueprint 当成独立服务。

11.3 误解：有了应用工厂就不会循环导入

应用工厂只能降低循环导入风险，不能自动消除所有循环依赖。

例如：

api.py 导入 service.py
service.py 导入 api.py

仍然会循环导入。正确做法是让依赖方向单向：

routes → service → repository → extension

而不是：

routes ↔ service

11.4 误解：`async def` 就获得异步并发

在 Flask 中：

async def view():
    ...

只表示视图函数是协程函数。它不会自动把 WSGI 服务器变成 ASGI 服务器，也不会让一个 worker 同时处理无限多个请求。Flask 官方明确指出，异步视图仍然占用一个 worker。(flask.palletsprojects.com)

11.5 典型错误及定位方法

错误一：访问 `current_app` 时报错

RuntimeError: Working outside of application context.

检查：

with app.app_context():
    function_using_current_app()

错误二：访问 `request` 时报错

RuntimeError: Working outside of request context.

检查调用是否真的发生在 HTTP 请求中；如果是测试，使用：

with app.test_request_context("/path"):
    ...

错误三：生产环境返回调试页面

检查是否启用了：

flask run --debug

或：

app.config["DEBUG"] = True

生产环境应关闭调试模式，并通过 WSGI 服务器启动。

错误四：生产环境修改代码却不生效

常见原因是：

使用了旧进程，没有重启；
多个 worker 只有部分重启；
容器镜像中仍是旧代码；
反向代理转发到了旧实例。

验证路径：

确认部署包版本
    ↓
确认 worker 进程启动时间
    ↓
确认反向代理 upstream
    ↓
访问 /healthz 或版本接口
    ↓
检查应用日志和服务器日志

可以在健康检查中暴露构建版本：

@app.get("/healthz")
def healthz():
    return {
        "status": "ok",
        "version": current_app.config["APP_VERSION"],
    }

不要在健康接口中执行昂贵的业务查询。存活检查、就绪检查和依赖检查应根据平台需求分别设计。

十二、如何选择 Flask 的工程形态

可以按应用特征选择：

直接全局应用

适合：

单文件脚本；
教学示例；
一次性内部工具；
不需要多环境测试的极小项目。

应用工厂 + Blueprint

适合：

中小型 API；
需要 pytest 测试；
需要开发、测试、生产多套配置；
需要拆分用户、订单、管理后台等模块。

Flask + WSGI 服务器

适合：

传统 HTTP API；
同步数据库驱动；
同步第三方 SDK；
以请求/响应为主的 Web 应用。

Flask + ASGI 适配器

适合：

已有 Flask 代码，希望接入 ASGI 基础设施；
需要逐步使用异步库；
不希望立即迁移到 ASGI 原生框架。

但它不能消除 Flask、WSGI 扩展和同步代码的限制。

ASGI 原生框架

适合：

WebSocket；
大量长连接；
以异步 I/O 为主；
需要原生 ASGI 中间件和生命周期；
希望异步并发成为框架的基础模型。

核心判断不是“哪个框架更先进”，而是应用的调用协议、依赖库、连接生命周期和并发模型是否一致。

Flask 的基础可以归结为一条清晰的数据流：

工厂创建应用
    ↓
扩展绑定应用
    ↓
Blueprint 注册路由
    ↓
服务器调用 WSGI 应用
    ↓
Flask 推入 Context
    ↓
请求钩子与视图函数运行
    ↓
错误处理与响应转换
    ↓
teardown 释放资源
    ↓
WSGI 服务器返回 HTTP 响应

理解这条链路后，current_app 为什么需要 Context、Blueprint 为什么必须注册、扩展为什么使用 init_app、开发服务器为什么不能用于生产，以及 Flask 的异步能力为什么不同于 ASGI，都会变成同一个生命周期模型下的自然推论。

系列导航与关联阅读

官方资料

本文依据 Python 官方文档、相关 PEP 与生态项目官方文档重新梳理；正文、示例与工程清单由 WR BLOG 编写。

Flask 完整基础：应用工厂、Context、Blueprint、扩展和部署

一、先建立 Flask 的整体模型

二、应用工厂：把“创建应用”变成函数

2.1 什么是应用工厂

2.2 直接全局创建应用的问题

2.3 一个更完整的配置加载顺序

2.4 工厂必须保持“可重复创建”

三、Context：为什么视图函数可以直接使用 request

3.1 Context 的问题背景

3.2 两种 Context

Application Context

Request Context

3.3 为什么应用上下文和请求上下文要分开

3.4 上下文外访问为什么报错

3.5 g 的正确用途：一次请求内复用资源

3.6 生命周期与异常

四、Blueprint：应用内部的模块化注册机制

4.1 Blueprint 是什么

4.2 推荐的目录结构

4.3 为什么 Blueprint 中使用 current_app

4.4 Blueprint 级别的钩子和错误处理

4.5 Blueprint 与 URL、端点名称

五、扩展：延迟绑定应用，而不是绑定某个全局应用

5.1 扩展的两阶段初始化

5.2 扩展初始化顺序

5.3 扩展不能替代业务边界

六、错误处理：把异常转换成稳定的 HTTP 契约

6.1 视图返回值如何变成响应

6.2 为 API 统一错误格式

6.3 错误处理器与 teardown 的关系

七、完整示例：工厂、Blueprint、扩展和数据库连接

7.1 项目结构

7.2 pyproject.toml

7.3 demo/db.py

7.4 demo/api.py

7.5 demo/__init__.py

八、测试：应用工厂的直接收益

九、WSGI、同步并发与 Flask 的异步视图

9.1 WSGI 的并发模型

9.2 Flask 的 async def 不等于 ASGI

9.3 WSGI 与 ASGI 的差异

9.4 Flask 运行在 ASGI 服务器上

十、生产部署：从 Flask CLI 到 WSGI 服务器

10.1 本地开发

10.2 使用 Waitress

10.3 使用 Gunicorn

10.4 反向代理的职责

10.5 进程、线程和数据库连接

十一、常见误解与诊断路径

11.1 误解：g 是全局缓存

11.2 误解：Blueprint 是可独立部署的子应用

11.3 误解：有了应用工厂就不会循环导入

11.4 误解：async def 就获得异步并发

11.5 典型错误及定位方法

错误一：访问 current_app 时报错

错误二：访问 request 时报错

错误三：生产环境返回调试页面

错误四：生产环境修改代码却不生效

十二、如何选择 Flask 的工程形态

直接全局应用

应用工厂 + Blueprint

Flask + WSGI 服务器

Flask + ASGI 适配器

ASGI 原生框架

系列导航与关联阅读

官方资料

三、Context：为什么视图函数可以直接使用 `request`

3.5 `g` 的正确用途：一次请求内复用资源

4.3 为什么 Blueprint 中使用 `current_app`

7.2 `pyproject.toml`

7.3 `demo/db.py`

7.4 `demo/api.py`

7.5 `demo/init.py`

9.2 Flask 的 `async def` 不等于 ASGI

11.1 误解：`g` 是全局缓存

11.4 误解：`async def` 就获得异步并发

错误一：访问 `current_app` 时报错

错误二：访问 `request` 时报错