flask

Flask 技能

经过生产环境验证的 Flask 模式，采用应用工厂模式、Blueprints 和 Flask-SQLAlchemy。

最新版本（于 2026 年 1 月验证）：

• Flask：3.1.2
• Flask-SQLAlchemy：3.1.1
• Flask-Login：0.6.3
• Flask-WTF：1.2.2
• Werkzeug：3.1.5
• Python：需要 3.9+（Flask 3.1.0 已弃用 3.8）

---

快速开始

使用 uv 设置项目

# Create project
uv init my-flask-app
cd my-flask-app

# Add dependencies
uv add flask flask-sqlalchemy flask-login flask-wtf python-dotenv

# Run development server
uv run flask --app app run --debug

最小可运行示例

# app.py
from flask import Flask

app = Flask(__name__)

@app.route("/")
def hello():
    return {"message": "Hello, World!"}

if __name__ == "__main__":
    app.run(debug=True)

运行：uv run flask --app app run --debug

---

已知问题预防

此技能可避免 9 个已记录的问题：

问题 #1：stream_with_context 清理回归问题（Flask 3.1.2）

错误：使用 stream_with_context 时，清理函数中出现 KeyError 来源：GitHub Issue #5804 发生原因：Flask 3.1.2 引入了一项回归问题，stream_with_context 会在响应生成完成前多次触发 teardown_request() 调用。如果清理回调使用不带默认值的 g.pop(key)，第二次调用时就会失败。

预防方法：

# WRONG - fails on second teardown call
@app.teardown_request
def _teardown_request(_):
    g.pop("hello")  # KeyError on second call

# RIGHT - idempotent teardown
@app.teardown_request
def _teardown_request(_):
    g.pop("hello", None)  # Provide default value

状态：该问题将作为 PR #5812 的副作用，在 Flask 3.2.0 中修复。在此之前，请确保所有清理回调都是幂等的。

---

问题 #2：异步视图与 Gevent 不兼容

错误：使用 gevent 处理并发异步请求时出现 RuntimeError 来源：GitHub Issue #5881 发生原因：激活 gevent 猴子补丁后，Asgiref 会失败。Asyncio 要求每个 OS 线程只有一个事件循环，但 gevent 的猴子补丁会让 threading.Thread 创建 greenlet，而不是真实线程，导致两个事件循环在同一个物理线程上运行并互相阻塞。

预防方法：选择异步（使用 asyncio/uvloop）或 gevent，不要同时使用。如果必须同时使用：

import asyncio
import gevent.monkey
import gevent.selectors
from flask import Flask

gevent.monkey.patch_all()
loop = asyncio.EventLoop(gevent.selectors.DefaultSelector())
gevent.spawn(loop.run_forever)

class GeventFlask(Flask):
    def async_to_sync(self, func):
        def run(*args, **kwargs):
            coro = func(*args, **kwargs)
            future = asyncio.run_coroutine_threadsafe(coro, loop)
            return future.result()
        return run

app = GeventFlask(__name__)

注意：维护者评论称，这会“彻底破坏两者的用途”。单个异步请求可以正常工作，但如果没有此变通方法，并发请求会失败。

---

问题 #3：重定向后测试客户端会话未更新

错误：测试中设置 follow_redirects=True 后，会话状态不正确 来源：GitHub Issue #5786 发生原因：在 Flask < 3.1.2 中，测试客户端跟随重定向后，会话未能正确更新。

预防方法：

# If using Flask >= 3.1.2, follow_redirects works correctly
def test_login_redirect(client):
    response = client.post('/login',
        data={'email': '[email protected]', 'password': 'pass'},
        follow_redirects=True)
    assert 'user_id' in session  # Works in 3.1.2+

# For Flask < 3.1.2, make separate requests
response = client.post('/login', data={...})
assert response.status_code == 302
response = client.get(response.location)  # Explicit redirect follow

状态：已在 Flask 3.1.2 中修复。请升级到最新版本。

---

问题 #4：线程中丢失应用上下文（社区来源）

错误：后台线程中出现 RuntimeError: Working outside of application context 来源：Sentry.io Guide 发生原因：将 current_app 传递给新线程时，必须使用 _get_current_object() 解包代理对象，并在线程中推送应用上下文。

预防方法：

from flask import current_app
import threading

# WRONG - current_app is a proxy, loses context in thread
def background_task():
    app_name = current_app.name  # Fails!

@app.route('/start')
def start_task():
    thread = threading.Thread(target=background_task)
    thread.start()

# RIGHT - unwrap proxy and push context
def background_task(app):
    with app.app_context():
        app_name = app.name  # Works!

@app.route('/start')
def start_task():
    app = current_app._get_current_object()
    thread = threading.Thread(target=background_task, args=(app,))
    thread.start()

已验证：这是生产应用中的常见模式，已记载于 Flask 官方文档。

---

问题 #5：Flask-Login 会话保护导致意外退出登录（社区来源）

错误：IP 地址变更时，用户意外退出登录 来源：Flask-Login Docs 发生原因：Flask-Login 的“strong”会话保护模式会在会话标识符（例如 IP 地址）变更时删除整个会话。这会影响使用移动网络或 VPN 的用户。

预防方法：

# app/extensions.py
from flask_login import LoginManager

login_manager = LoginManager()
login_manager.session_protection = "basic"  # Default, less strict
# login_manager.session_protection = "strong"  # Strict, may logout on IP change
# login_manager.session_protection = None  # Disabled (not recommended)

注意：默认情况下，Flask-Login 允许并发会话（同一用户在多个浏览器上登录）。如需阻止此行为，请实现自定义会话跟踪。

已验证：Flask-Login 官方文档及多篇 2024 年博客文章。

---

Issue #6：CSRF 保护与缓存冲突（社区提供）

错误：缓存页面上的表单提交失败，并提示“缺少/无效的 CSRF 令牌” 来源：Flask-WTF Docs 发生原因：如果 Web 服务器的缓存策略将页面缓存的时间设置得长于 WTF_CSRF_TIME_LIMIT，浏览器就会提供带有已过期 CSRF 令牌的缓存页面。

预防措施：

# Option 1: Align cache duration with token lifetime
WTF_CSRF_TIME_LIMIT = None  # Never expire (less secure)

# Option 2: Exclude forms from cache
@app.after_request
def add_cache_headers(response):
    if request.method == 'GET' and 'form' in request.endpoint:
        response.headers['Cache-Control'] = 'no-cache, no-store, must-revalidate'
    return response

# Option 3: Configure webserver to not cache POST targets
# In Nginx: add "proxy_cache_bypass $cookie_session" for form routes

已验证：Flask-WTF 官方文档警告，以及 2024 年的安全最佳实践指南。

---

Issue #7：按请求覆盖 max_content_length（新功能）

功能：Flask 3.1.0 新增了按请求自定义 Request.max_content_length 的功能 来源：Flask 3.1.0 Release Notes

用法：

from flask import Flask, request

app = Flask(__name__)
app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024  # 16MB default

@app.route('/upload', methods=['POST'])
def upload():
    # Override for this specific route
    request.max_content_length = 100 * 1024 * 1024  # 100MB for uploads
    file = request.files['file']
    # ...

注意：3.1.0 还新增了 MAX_FORM_MEMORY_SIZE 和 MAX_FORM_PARTS 配置选项。请参阅安全文档。

---

Issue #8：SECRET_KEY 轮换（新功能）

功能：Flask 3.1.0 新增了用于密钥轮换的 SECRET_KEY_FALLBACKS 来源：Flask 3.1.0 Release Notes

用法：

# config.py
class Config:
    SECRET_KEY = "new-secret-key-2024"
    SECRET_KEY_FALLBACKS = [
        "old-secret-key-2023",
        "older-secret-key-2022"
    ]

注意：扩展需要明确支持此功能。Flask-Login 和 Flask-WTF 可能需要更新才能使用备用密钥。

---

Issue #9：Werkzeug 3.1+ 依赖冲突

错误：flask==2.2.4 incompatible with werkzeug==3.1.3 来源：Flask 3.1.0 Release Notes | GitHub Issue #5652 发生原因：Flask 3.1.0 更新了最低依赖版本：Werkzeug >= 3.1、ItsDangerous >= 2.2、Blinker >= 1.9。固定使用旧版本的项目会发生冲突。

预防措施：

# Update all Pallets projects together
pip install flask>=3.1.0 werkzeug>=3.1.0 itsdangerous>=2.2.0 blinker>=1.9.0

# Or with uv
uv add "flask>=3.1.0" "werkzeug>=3.1.0" "itsdangerous>=2.2.0" "blinker>=1.9.0"

---

项目结构（应用工厂）

要构建易于维护的应用，请使用带蓝图的工厂模式：

my-flask-app/
├── pyproject.toml
├── config.py                # Configuration classes
├── run.py                   # Entry point
│
├── app/
│   ├── __init__.py          # Application factory (create_app)
│   ├── extensions.py        # Flask extensions (db, login_manager)
│   ├── models.py            # SQLAlchemy models
│   │
│   ├── main/                # Main blueprint
│   │   ├── __init__.py
│   │   └── routes.py
│   │
│   ├── auth/                # Auth blueprint
│   │   ├── __init__.py
│   │   ├── routes.py
│   │   └── forms.py
│   │
│   ├── templates/
│   │   ├── base.html
│   │   ├── main/
│   │   └── auth/
│   │
│   └── static/
│       ├── css/
│       └── js/
│
└── tests/
    ├── conftest.py
    └── test_main.py

---

核心模式

应用工厂

# app/__init__.py
from flask import Flask
from app.extensions import db, login_manager
from config import Config


def create_app(config_class=Config):
    """Application factory function."""
    app = Flask(__name__)
    app.config.from_object(config_class)

    # Initialize extensions
    db.init_app(app)
    login_manager.init_app(app)

    # Register blueprints
    from app.main import bp as main_bp
    from app.auth import bp as auth_bp

    app.register_blueprint(main_bp)
    app.register_blueprint(auth_bp, url_prefix="/auth")

    # Create database tables
    with app.app_context():
        db.create_all()

    return app

主要优势：

• 可使用不同配置创建多个应用实例（用于测试）
• 避免循环导入
• 扩展只初始化一次，之后再绑定到应用

扩展模块

# app/extensions.py
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager

db = SQLAlchemy()
login_manager = LoginManager()
login_manager.login_view = "auth.login"
login_manager.login_message_category = "info"

为何使用单独的文件？：可避免循环导入，让模型能够导入 db 而无需导入 app。

配置

# config.py
import os
from dotenv import load_dotenv

load_dotenv()


class Config:
    """Base configuration."""
    SECRET_KEY = os.environ.get("SECRET_KEY", "dev-secret-key")
    SQLALCHEMY_DATABASE_URI = os.environ.get("DATABASE_URL", "sqlite:///app.db")
    SQLALCHEMY_TRACK_MODIFICATIONS = False


class DevelopmentConfig(Config):
    """Development configuration."""
    DEBUG = True


class TestingConfig(Config):
    """Testing configuration."""
    TESTING = True
    SQLALCHEMY_DATABASE_URI = "sqlite:///:memory:"
    WTF_CSRF_ENABLED = False


class ProductionConfig(Config):
    """Production configuration."""
    DEBUG = False

入口点

# run.py
from app import create_app

app = create_app()

if __name__ == "__main__":
    app.run()

运行：flask --app run run --debug

---

蓝图

创建蓝图

# app/main/__init__.py
from flask import Blueprint

bp = Blueprint("main", __name__)

from app.main import routes  # Import routes after bp is created!

# app/main/routes.py
from flask import render_template, jsonify
from app.main import bp


@bp.route("/")
def index():
    return render_template("main/index.html")


@bp.route("/api/health")
def health():
    return jsonify({"status": "ok"})

带模板的蓝图

# app/auth/__init__.py
from flask import Blueprint

bp = Blueprint(
    "auth",
    __name__,
    template_folder="templates",  # Blueprint-specific templates
    static_folder="static",       # Blueprint-specific static files
)

from app.auth import routes

---

数据库模型

# app/models.py
from datetime import datetime
from flask_login import UserMixin
from werkzeug.security import generate_password_hash, check_password_hash
from app.extensions import db, login_manager


class User(UserMixin, db.Model):
    """User model for authentication."""
    __tablename__ = "users"

    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String(120), unique=True, nullable=False, index=True)
    password_hash = db.Column(db.String(256), nullable=False)
    is_active = db.Column(db.Boolean, default=True)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)

    def set_password(self, password):
        self.password_hash = generate_password_hash(password)

    def check_password(self, password):
        return check_password_hash(self.password_hash, password)

    def __repr__(self):
        return f"<User {self.email}>"


@login_manager.user_loader
def load_user(user_id):
    return User.query.get(int(user_id))

---

使用 Flask-Login 进行身份验证

身份验证表单

# app/auth/forms.py
from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, BooleanField, SubmitField
from wtforms.validators import DataRequired, Email, Length, EqualTo, ValidationError
from app.models import User


class LoginForm(FlaskForm):
    email = StringField("Email", validators=[DataRequired(), Email()])
    password = PasswordField("Password", validators=[DataRequired()])
    remember = BooleanField("Remember Me")
    submit = SubmitField("Login")


class RegistrationForm(FlaskForm):
    email = StringField("Email", validators=[DataRequired(), Email()])
    password = PasswordField("Password", validators=[DataRequired(), Length(min=8)])
    confirm = PasswordField("Confirm Password", validators=[
        DataRequired(), EqualTo("password", message="Passwords must match")
    ])
    submit = SubmitField("Register")

    def validate_email(self, field):
        if User.query.filter_by(email=field.data).first():
            raise ValidationError("Email already registered.")

身份验证路由

# app/auth/routes.py
from flask import render_template, redirect, url_for, flash, request
from flask_login import login_user, logout_user, login_required, current_user
from app.auth import bp
from app.auth.forms import LoginForm, RegistrationForm
from app.extensions import db
from app.models import User


@bp.route("/register", methods=["GET", "POST"])
def register():
    if current_user.is_authenticated:
        return redirect(url_for("main.index"))

    form = RegistrationForm()
    if form.validate_on_submit():
        user = User(email=form.email.data)
        user.set_password(form.password.data)
        db.session.add(user)
        db.session.commit()
        flash("Registration successful! Please log in.", "success")
        return redirect(url_for("auth.login"))

    return render_template("auth/register.html", form=form)


@bp.route("/login", methods=["GET", "POST"])
def login():
    if current_user.is_authenticated:
        return redirect(url_for("main.index"))

    form = LoginForm()
    if form.validate_on_submit():
        user = User.query.filter_by(email=form.email.data).first()
        if user and user.check_password(form.password.data):
            login_user(user, remember=form.remember.data)
            next_page = request.args.get("next")
            flash("Logged in successfully!", "success")
            return redirect(next_page or url_for("main.index"))
        flash("Invalid email or password.", "danger")

    return render_template("auth/login.html", form=form)


@bp.route("/logout")
@login_required
def logout():
    logout_user()
    flash("You have been logged out.", "info")
    return redirect(url_for("main.index"))

保护路由

from flask_login import login_required, current_user

@bp.route("/dashboard")
@login_required
def dashboard():
    return render_template("main/dashboard.html", user=current_user)

---

API 路由（JSON）

用于不使用模板的 REST API：

# app/api/__init__.py
from flask import Blueprint

bp = Blueprint("api", __name__)

from app.api import routes

# app/api/routes.py
from flask import jsonify, request
from flask_login import login_required, current_user
from app.api import bp
from app.extensions import db
from app.models import User


@bp.route("/users", methods=["GET"])
@login_required
def get_users():
    users = User.query.all()
    return jsonify([
        {"id": u.id, "email": u.email}
        for u in users
    ])


@bp.route("/users", methods=["POST"])
def create_user():
    data = request.get_json()
    if not data or "email" not in data or "password" not in data:
        return jsonify({"error": "Missing required fields"}), 400

    if User.query.filter_by(email=data["email"]).first():
        return jsonify({"error": "Email already exists"}), 409

    user = User(email=data["email"])
    user.set_password(data["password"])
    db.session.add(user)
    db.session.commit()

    return jsonify({"id": user.id, "email": user.email}), 201

使用前缀注册：

app.register_blueprint(api_bp, url_prefix="/api/v1")

---

关键规则

务必做到

1. 使用应用工厂模式 - 支持测试，避免全局变量
2. 将扩展放在单独的文件中 - 避免循环导入
3. 在蓝图 __init__.py 的底部导入路由 - 在创建 bp 之后
4. 使用 current_app，而不是 app - 在请求上下文中
5. 使用 with app.app_context() - 在请求之外访问 db 时

切勿这样做

1. 切勿在模型中导入 app - 会导致循环导入
2. 切勿在应用上下文之外访问 db - 会引发 RuntimeError
3. 切勿将密钥存储在代码中 - 使用环境变量
4. 切勿在生产环境中使用 app.run() - 使用 Gunicorn
5. 切勿跳过 CSRF 保护 - 保持启用 Flask-WTF

---

常见错误与修复方法

循环导入错误

错误：ImportError: cannot import name 'X' from partially initialized module

原因：模型导入 app，而 app 导入模型

修复方法：使用 extensions.py 模式：

# WRONG - circular import
# app/__init__.py
from app.models import User  # models.py imports db from here!

# RIGHT - deferred import
# app/__init__.py
def create_app():
    # ... setup ...
    from app.models import User  # Import inside factory

在应用上下文之外工作

错误：RuntimeError: Working outside of application context

原因：在请求之外访问 current_app、g 或 db

修复方法：

# WRONG
from app import create_app
app = create_app()
users = User.query.all()  # No context!

# RIGHT
from app import create_app
app = create_app()
with app.app_context():
    users = User.query.all()  # Has context

未找到蓝图

错误：werkzeug.routing.BuildError: Could not build url for endpoint

原因：在 url_for() 中使用了错误的蓝图前缀

修复：

# WRONG
url_for("login")

# RIGHT - include blueprint name
url_for("auth.login")

缺少 CSRF 令牌

错误：Bad Request: The CSRF token is missing

原因：提交的表单未包含 CSRF 令牌

修复：在模板中包含令牌：

<form method="post">
    {{ form.hidden_tag() }}  
    
</form>

---

测试

# tests/conftest.py
import pytest
from app import create_app
from app.extensions import db
from config import TestingConfig


@pytest.fixture
def app():
    app = create_app(TestingConfig)
    with app.app_context():
        db.create_all()
        yield app
        db.drop_all()


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


@pytest.fixture
def runner(app):
    return app.test_cli_runner()

# tests/test_main.py
def test_index(client):
    response = client.get("/")
    assert response.status_code == 200


def test_register(client):
    response = client.post("/auth/register", data={
        "email": "[email protected]",
        "password": "testpass123",
        "confirm": "testpass123",
    }, follow_redirects=True)
    assert response.status_code == 200

运行：uv run pytest

---

部署

开发环境

flask --app run run --debug

使用 Gunicorn 部署到生产环境

uv add gunicorn
uv run gunicorn -w 4 -b 0.0.0.0:8000 "run:app"

Docker

FROM python:3.12-slim

WORKDIR /app
COPY . .

RUN pip install uv && uv sync

EXPOSE 8000
CMD ["uv", "run", "gunicorn", "-w", "4", "-b", "0.0.0.0:8000", "run:app"]

环境变量（.env）

SECRET_KEY=your-production-secret-key
DATABASE_URL=postgresql://user:pass@localhost/dbname
FLASK_ENV=production

---

参考资料

• Flask 文档
• Flask-SQLAlchemy
• Flask-Login
• Flask Mega-Tutorial
• 应用工厂模式

---

最后验证时间：2026-01-21 | 技能版本：2.0.0 | 变更：新增 9 个已知问题（stream_with_context 回归、async/gevent 冲突、测试客户端会话、线程上下文、Flask-Login 会话保护、CSRF 缓存、新增 3.1.0 功能、Werkzeug 依赖） 维护者：Jezweb | [email protected]