Flask

技术栈
后端框架
pythonweb框架微框架WSGIJinja2RESTful

概览

Flask 技术栈概览

Flask 是一个用 Python 编写的轻量级、微框架的 Web 应用框架,由 Armin Ronacher 于 2010 年发布。它基于 Werkzeug WSGI 工具包和 Jinja2 模板引擎,遵循"微内核 + 插件"的设计哲学——核心极简,只提供路由、请求/响应处理和模板渲染等基础能力,其他功能(ORM、表单验证、认证等)通过丰富的扩展生态按需集成。

核心特性:

  • 🎯 微框架设计 — 核心小巧精悍,按需扩展,不强制任何项目结构
  • 🔌 丰富的扩展生态 — Flask-SQLAlchemy、Flask-Login、Flask-Migrate、Flask-RESTful 等
  • 🧩 Jinja2 模板引擎 — 强大的模板继承、宏、过滤器系统
  • 🔄 RESTful 友好 — 内置 URL 路由支持多种 HTTP 方法和请求数据解析
  • 🧪 内置测试客户端 — 无需额外工具即可进行集成测试
  • 📦 单文件应用 — 最小应用仅需 7 行代码即可运行

适用场景: 中小型 Web 应用、RESTful API 服务、微服务、原型快速验证、静态站点生成。

安装

1. 环境准备

  • 操作系统: Windows 10+ / macOS 11+ / Linux(Ubuntu 20.04+、Debian 11+、CentOS 8+)
  • Python 版本: Python 3.9 及以上(推荐 3.11+)
  • 依赖项: pip(Python 包管理器,通常随 Python 一起安装)、virtualenv 或 venv(虚拟环境工具,Python 3.3+ 内置)

创建虚拟环境(强烈推荐)

# 创建虚拟环境
python -m venv venv

# 激活虚拟环境
# Windows:
venv\Scripts\activate
# macOS / Linux:
source venv/bin/activate

2. 安装命令

# 安装 Flask
pip install flask

# 安装常用扩展
pip install flask-sqlalchemy   # ORM 数据库支持
pip install flask-migrate      # 数据库迁移
pip install flask-login        # 用户认证
pip install flask-wtf          # 表单与 CSRF 保护
pip install flask-restful      # REST API 构建
pip install flask-cors         # 跨域支持

# 一键安装所有开发依赖
pip install flask flask-sqlalchemy flask-migrate flask-login flask-wtf flask-restful flask-cors python-dotenv

# 验证安装
flask --version

3. 常见安装问题

问题 1:pip: command not found

# macOS 使用 Homebrew
brew install python3
# Ubuntu/Debian
sudo apt update &;& sudo apt install python3-pip
# 或者使用 python3 -m pip 代替 pip
python3 -m pip install flask

问题 2:虚拟环境激活失败(Windows PowerShell 权限问题)

# 以管理员身份运行 PowerShell,然后执行
Set-ExecutionPolicy RemoteSigned -Scope CurrentUser
# 再激活
venv\Scripts\Activate.ps1

问题 3:ImportError: No module named flask

确保虚拟环境已激活,终端前缀应显示 (venv)。如果仍报错,重新安装:

pip uninstall flask
pip install flask

问题 4:安装速度慢

# 使用国内镜像源(清华源)
pip install flask -i https://pypi.tuna.tsinghua.edu.cn/simple
# 或永久配置
pip config set global.index-url https://pypi.tuna.tsinghua.edu.cn/simple

问题 5:端口 5000 被占用(macOS Monterey+ AirPlay)

macOS 自 Monterey 起 AirPlay Receiver 占用 5000 端口,解决方法:

# 方法一:更换端口
flask run --port 8080

# 方法二:关闭 AirPlay Receiver(系统设置 → 通用 → AirDrop 与 Handoff → 关闭 AirPlay 接收器)

示例

Flask Hello World —— 你的第一个 Flask 应用

目标

创建一个最小的 Flask Web 应用,理解:

  • Flask 应用实例的创建
  • 路由装饰器的使用
  • 请求-响应基本流程
  • 开发服务器的启动

完整代码

# app.py
from flask import Flask, request, jsonify

# 创建 Flask 应用实例
app = Flask(__name__)

# ---------- 路由定义 ----------

@app.route("/")
def index():
    """首页 —— 返回 HTML"""
    return """
    <h1>🚀 Hello, Flask!</h1>
    <p>欢迎来到 Flask 世界。</p>
    <ul>
        <li><a href="/hello">/hello</a> — 纯文本问候</li>
        <li><a href="/hello?name=Vibe">/hello?name=Vibe</a> — 带参数问候</li>
        <li><a href="/api/status">/api/status</a> — JSON 状态接口</li>
    </ul>
    """


@app.route("/hello")
def hello():
    """纯文本问候,支持查询参数 name"""
    name = request.args.get("name", "World")
    return f"Hello, {name}!"


@app.route("/api/status")
def api_status():
    """RESTful JSON 状态接口"""
    return jsonify({
        "status": "ok",
        "version": "1.0.0",
        "framework": "Flask"
    })


@app.route("/user/<username>")
def user_profile(username):
    """动态路由 —— URL 变量"""
    return jsonify({
        "username": username,
        "message": f"欢迎你,{username}!"
    })


# ---------- 错误处理 ----------

@app.errorhandler(404)
def not_found(error):
    return jsonify({"error": "资源未找到"}), 404


# ---------- 启动入口 ----------
if __name__ == "__main__":
    # debug=True 开启热重载,代码修改后自动重启
    app.run(host="0.0.0.0", port=5000, debug=True)

运行步骤

# 1. 确保虚拟环境已激活(终端前缀为 (venv))
# 2. 设置环境变量(也可在代码中直接 run)
# Linux/macOS:
export FLASK_APP=app.py
export FLASK_ENV=development
# Windows (CMD):
set FLASK_APP=app.py
set FLASK_ENV=development
# Windows (PowerShell):
$env:FLASK_APP="app.py"
$env:FLASK_ENV="development"

# 3. 启动
flask run
# 或者直接运行
python app.py

# 4. 浏览器访问
# http://localhost:5000/
# http://localhost:5000/hello?name=Vibe
# http://localhost:5000/api/status
# http://localhost:5000/user/alice

预期输出

访问 http://localhost:5000/api/status 返回:

{
  "status": "ok",
  "version": "1.0.0",
  "framework": "Flask"
}

访问 http://localhost:5000/hello?name=Vibe 返回:

Hello, Vibe!

关键要点

概念 说明
Flask(__name__) 创建 WSGI 应用实例,__name__ 用于定位资源
@app.route(path) 装饰器语法,将 URL 路径绑定到视图函数
<variable> 动态路由变量,自动传入函数参数
request.args 查询参数字典 ?key=value
jsonify() 将 dict 转为 JSON 响应,并设 Content-Type
debug=True 开启调试模式,代码修改后自动重载

Flask RESTful API —— 构建完整的 CRUD 接口

目标

使用 Flask 构建一个完整的 RESTful API,实现对"书籍"资源的 CRUD 操作:

  • 理解 RESTful URL 设计规范
  • 掌握请求体 JSON 解析
  • 掌握状态码的正确使用
  • 使用 Flask-RESTful 扩展简化 API 开发

完整代码

# app_rest.py
from flask import Flask, request, jsonify, abort
from flask_restful import Api, Resource, reqparse, fields, marshal_with
from datetime import datetime

app = Flask(__name__)
api = Api(app)

# ---------- 模拟数据库 ----------
books_db = [
    {"id": 1, "title": "1984", "author": "George Orwell", "year": 1949, "created_at": "2024-01-01T00:00:00"},
    {"id": 2, "title": "三体", "author": "刘慈欣", "year": 2008, "created_at": "2024-01-02T00:00:00"},
]

# ---------- 请求解析器(输入验证)----------
book_parser = reqparse.RequestParser()
book_parser.add_argument("title", type=str, required=True, help="书名不能为空")
book_parser.add_argument("author", type=str, required=True, help="作者不能为空")
book_parser.add_argument("year", type=int, required=True, help="出版年份必须为整数")

# ---------- 响应格式化 ----------
book_fields = {
    "id": fields.Integer,
    "title": fields.String,
    "author": fields.String,
    "year": fields.Integer,
    "created_at": fields.String,
}


class BookListResource(Resource):
    """处理 /api/books 的 GET(列表)和 POST(创建)"""

    def get(self):
        """获取所有书籍"""
        return {"books": books_db, "total": len(books_db)}

    @marshal_with(book_fields)
    def post(self):
        """创建新书籍"""
        args = book_parser.parse_args()
        new_id = max(b["id"] for b in books_db) + 1 if books_db else 1
        new_book = {
            "id": new_id,
            "title": args["title"],
            "author": args["author"],
            "year": args["year"],
            "created_at": datetime.now().isoformat(),
        }
        books_db.append(new_book)
        return new_book, 201


class BookResource(Resource):
    """处理 /api/books/<book_id> 的 GET / PUT / DELETE"""

    @marshal_with(book_fields)
    def get(self, book_id):
        """获取单本书"""
        book = next((b for b in books_db if b["id"] == book_id), None)
        if book is None:
            abort(404, message=f"书籍 {book_id} 不存在")
        return book

    @marshal_with(book_fields)
    def put(self, book_id):
        """更新书籍"""
        book = next((b for b in books_db if b["id"] == book_id), None)
        if book is None:
            abort(404, message=f"书籍 {book_id} 不存在")
        args = book_parser.parse_args()
        book["title"] = args["title"]
        book["author"] = args["author"]
        book["year"] = args["year"]
        return book

    def delete(self, book_id):
        """删除书籍"""
        global books_db
        book = next((b for b in books_db if b["id"] == book_id), None)
        if book is None:
            abort(404, message=f"书籍 {book_id} 不存在")
        books_db = [b for b in books_db if b["id"] != book_id]
        return {"message": f"书籍 {book_id} 已删除"}, 200


# ---------- 注册路由 ----------
api.add_resource(BookListResource, "/api/books")
api.add_resource(BookResource, "/api/books/<int:book_id>")


# ---------- 错误处理 ----------
@app.errorhandler(404)
def handle_404(error):
    return jsonify({"error": str(error)}), 404


@app.errorhandler(400)
def handle_400(error):
    return jsonify({"error": str(error)}), 400


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000, debug=True)

测试 API

# 获取所有书籍
curl http://localhost:5000/api/books

# 获取单本书
curl http://localhost:5000/api/books/1

# 创建新书
curl -X POST http://localhost:5000/api/books \
  -H "Content-Type: application/json" \
  -d '{"title":"Animal Farm","author":"George Orwell","year":1945}'

# 更新书籍
curl -X PUT http://localhost:5000/api/books/1 \
  -H "Content-Type: application/json" \
  -d '{"title":"Nineteen Eighty-Four","author":"George Orwell","year":1949}'

# 删除书籍
curl -X DELETE http://localhost:5000/api/books/2

RESTful URL 设计规范

方法 URL 操作 状态码
GET /api/books 获取列表 200
GET /api/books/1 获取单个 200 / 404
POST /api/books 创建 201
PUT /api/books/1 更新 200 / 404
DELETE /api/books/1 删除 200 / 404

关键要点

  • reqparse:用于请求参数验证,自动返回 400 错误
  • marshal_with:装饰器自动将返回 dict 按 book_fields 格式化
  • abort():中断请求并返回错误响应
  • 类视图 + add_resource:比装饰器更符合 RESTful 组织方式

教程

Flask 入门教程 —— 从零构建 Web 应用

本章目标

  • 理解 Flask 的核心设计理念
  • 搭建完整的 Flask 项目骨架
  • 掌握路由、模板、静态文件三大基础
  • 实现一个简单的留言板应用

1. Flask 的设计哲学

Flask 遵循微内核 + 扩展的设计模式:

Flask 核心(最小集合)
├── 路由系统(werkzeug.routing)
├── 请求/响应封装(werkzeug.wrappers)
├── 模板渲染(Jinja2)
├── 开发服务器
└── 测试客户端

按需扩展(pip install 自由搭配)
├── flask-sqlalchemy    → 数据库
├── flask-login         → 认证
├── flask-admin         → 后台管理
├── flask-socketio      → WebSocket
└── ...上百个扩展

对比 Django:Django 是"大而全"的框架,自带 ORM、后台、表单、认证;Flask 是一张白纸,你可以自由选择每个组件。

2. 项目骨架

my_flask_app/
├── app/                    # 应用包
│   ├── __init__.py         # 工厂函数 create_app()
│   ├── routes.py           # 路由视图
│   ├── models.py           # 数据库模型
│   ├── forms.py            # 表单定义
│   ├── templates/          # Jinja2 模板
│   │   ├── base.html
│   │   ├── index.html
│   │   └── post.html
│   └── static/             # 静态文件
│       ├── css/
│       └── js/
├── migrations/             # 数据库迁移
├── config.py               # 配置类
├── requirements.txt        # 依赖
└── run.py                  # 启动入口

3. 应用程序工厂模式

# app/__init__.py
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate

db = SQLAlchemy()
migrate = Migrate()


def create_app():
    app = Flask(__name__)
    app.config.from_object("config.Config")

    # 初始化扩展
    db.init_app(app)
    migrate.init_app(app, db)

    # 注册蓝图
    from app.routes import main_bp
    app.register_blueprint(main_bp)

    return app

为什么用工厂模式?

  • 支持多环境配置(开发/测试/生产)
  • 便于单元测试(每次 create_app() 都是全新实例)
  • 延迟初始化,避免循环导入

4. 蓝图(Blueprint)组织路由

# app/routes.py
from flask import Blueprint, render_template, request, redirect, url_for
from app.models import Message
from app import db

main_bp = Blueprint("main", __name__)


@main_bp.route("/")
def index():
    messages = Message.query.order_by(Message.created_at.desc()).all()
    return render_template("index.html", messages=messages)


@main_bp.route("/post", methods=["POST"])
def post():
    content = request.form.get("content", "").strip()
    if content:
        msg = Message(content=content)
        db.session.add(msg)
        db.session.commit()
    return redirect(url_for("main.index"))

蓝图的好处

  • 将大型应用拆分为独立的"迷你应用"
  • 每个蓝图可有自己的模板/静态文件目录
  • 支持 URL 前缀(如 /admin/api/v1

5. Jinja2 模板实战

<!-- app/templates/base.html -->
<!DOCTYPE html>
<html lang="zh">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>{% block title %}留言板{% endblock %}</title>
    <link rel="stylesheet" href="{{ url_for('static', filename='css/style.css') }}">
</head>
<body>
    <div class="container">
        {% with messages = get_flashed_messages(with_categories=true) %}
            {% if messages %}
                {% for category, message in messages %}
                    <div class="alert alert-{{ category }}">{{ message }}</div>
                {% endfor %}
            {% endif %}
        {% endwith %}

        {% block content %}{% endblock %}
    </div>
</body>
</html>
<!-- app/templates/index.html -->
{% extends "base.html" %}
{% block title %}留言板 - 首页{% endblock %}

{% block content %}
<h1>📝 留言板</h1>

<form method="post" action="{{ url_for('main.post') }}">
    <textarea name="content" rows="3" placeholder="说点什么吧..." required></textarea>
    <button type="submit">发布</button>
</form>

<hr>

{% for msg in messages %}
    <div class="message">
        <p>{{ msg.content }}</p>
        <small>{{ msg.created_at.strftime('%Y-%m-%d %H:%M') }}</small>
    </div>
{% else %}
    <p>还没有留言,来写第一条吧!</p>
{% endfor %}
{% endblock %}

Jinja2 关键语法

语法 说明
{% block name %} 定义可被子模板替换的块
{% extends "base.html" %} 继承基础模板
{% for ... %} / {% endfor %} 循环
{% if ... %} / {% endif %} 条件判断
{% with ... %} 变量作用域
{{ variable }} 输出变量(自动 HTML 转义)
{{ | safe }} 不转义输出(慎用)
url_for() 反向生成 URL(路由改名时自动更新)

6. 使用 Flask-Login 实现用户认证

from flask_login import LoginManager, UserMixin, login_user, logout_user, login_required, current_user

login_manager = LoginManager()
login_manager.login_view = "auth.login"


class User(db.Model, UserMixin):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(64), unique=True, nullable=False)
    password_hash = db.Column(db.String(128), nullable=False)

    def set_password(self, password):
        from werkzeug.security import generate_password_hash
        self.password_hash = generate_password_hash(password)

    def check_password(self, password):
        from werkzeug.security import check_password_hash
        return check_password_hash(self.password_hash, password)


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


# 保护路由
@main_bp.route("/secret")
@login_required
def secret():
    return f"只有登录用户能看到,你好 {current_user.username}!"

7. 配置管理最佳实践

# config.py
import os
from dotenv import load_dotenv

load_dotenv()  # 加载 .env 文件


class Config:
    SECRET_KEY = os.getenv("SECRET_KEY", "dev-key-change-me")
    SQLALCHEMY_DATABASE_URI = os.getenv("DATABASE_URL", "sqlite:///app.db")
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    JSON_AS_ASCII = False  # 支持中文 JSON 输出


class DevelopmentConfig(Config):
    DEBUG = True


class ProductionConfig(Config):
    DEBUG = False


class TestingConfig(Config):
    TESTING = True
    SQLALCHEMY_DATABASE_URI = "sqlite:///:memory:"

思考题

  1. 蓝图与直接 @app.route 有何区别?何时应该使用蓝图?
  2. url_for() 比硬编码 URL 有哪些优势?
  3. Flask 的"微"体现在哪里?为什么说这是优势而非劣势?
  4. 工厂模式 create_app() 相比全局 app = Flask(__name__) 有什么好处?

Flask 进阶实战 —— 异步任务、Docker 部署与性能优化

本章目标

  • 掌握 Celery + Redis 异步任务队列
  • 使用 Docker 容器化 Flask 应用
  • 了解 Gunicorn 生产部署
  • 常见性能优化策略

1. Celery 异步任务

1.1 为什么需要异步任务?

Web 请求应该在毫秒级完成。以下场景必须异步处理:

  • 发送邮件验证码(可能耗时 2~5 秒)
  • 生成报表/PDF(可能耗时 10 秒+)
  • 图像/视频处理
  • 调用第三方 API(超时风险)

1.2 架构概览

用户请求 → Flask (快速返回 202 Accepted)
                │
                ↓
         Redis/RabbitMQ (消息队列/代理)
                │
                ↓
         Celery Worker (后台异步执行)
                │
                ↓
         Redis/DB (结果存储)
                │
                ↓
         用户轮询/WebSocket 获取结果

1.3 配置与使用

# tasks.py
from celery import Celery

celery = Celery(
    "my_flask_tasks",
    broker="redis://localhost:6379/0",       # 消息代理
    backend="redis://localhost:6379/1",      # 结果后端
)

# 配置序列化
celery.conf.update(
    task_serializer="json",
    accept_content=["json"],
    result_expires=3600,  # 结果过期时间(秒)
)


@celery.task(bind=True, max_retries=3)
def send_welcome_email(self, user_email, username):
    """发送欢迎邮件(带重试机制)"""
    try:
        # 模拟发送邮件
        import smtplib
        import time
        time.sleep(2)  # 模拟耗时操作
        print(f"✅ 已发送欢迎邮件至 {user_email}")
        return {"status": "ok", "email": user_email}
    except Exception as exc:
        print(f"❌ 发送失败,将重试: {exc}")
        raise self.retry(exc=exc, countdown=60)  # 60秒后重试


@celery.task
def generate_report(report_id):
    """生成报表——耗时任务"""
    import time
    time.sleep(5)  # 模拟复杂计算
    # ... 实际报表生成逻辑
    return {"report_id": report_id, "status": "done"}
# app/routes.py 中使用异步任务
from tasks import send_welcome_email, generate_report

@main_bp.route("/register", methods=["POST"])
def register():
    username = request.form["username"]
    email = request.form["email"]

    # 创建用户... (省略)
    
    # 异步发送欢迎邮件
    task = send_welcome_email.delay(email, username)
    
    return jsonify({
        "message": "注册成功,欢迎邮件正在发送中",
        "task_id": task.id,
    }), 202


@main_bp.route("/task/<task_id>")
def get_task_status(task_id):
    """查询异步任务状态"""
    from celery.result import AsyncResult
    result = AsyncResult(task_id, app=celery)
    
    if result.state == "PENDING":
        response = {"state": "PENDING", "progress": "等待执行..."}
    elif result.state == "SUCCESS":
        response = {"state": "SUCCESS", "result": result.get()}
    elif result.state == "FAILURE":
        response = {"state": "FAILURE", "error": str(result.info)}
    else:
        response = {"state": result.state}
    
    return jsonify(response)

1.4 启动命令

# 需要先安装并启动 Redis
sudo apt install redis-server  # Linux
brew install redis              # macOS

# 启动 Celery Worker
celery -A tasks.celery worker --loglevel=info --concurrency=4

# 启动 Celery Beat(定时任务调度,可选)
celery -A tasks.celery beat --loglevel=info

# 同时启动 Worker 和 Beat
celery -A tasks.celery worker -B --loglevel=info

2. Docker 容器化部署

# Dockerfile
FROM python:3.11-slim

# 设置工作目录
WORKDIR /app

# 安装系统依赖
RUN apt-get update &;& apt-get install -y --no-install-recommends \
    gcc \
    &;& rm -rf /var/lib/apt/lists/*

# 复制并安装 Python 依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple

# 复制应用代码
COPY . .

# 创建非 root 用户
RUN useradd -m flaskuser &;& chown -R flaskuser:flaskuser /app
USER flaskuser

# 暴露端口
EXPOSE 8000

# 生产环境使用 Gunicorn 启动
CMD ["gunicorn", "-w", "4", "-b", "0.0.0.0:8000", "run:app"]
# docker-compose.yml
version: "3.8"

services:
  web:
    build: .
    ports:
      - "8000:8000"
    environment:
      - FLASK_ENV=production
      - DATABASE_URL=postgresql://user:pass@db:5432/flaskdb
      - REDIS_URL=redis://redis:6379/0
    depends_on:
      - db
      - redis

  db:
    image: postgres:15-alpine
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
      POSTGRES_DB: flaskdb
    volumes:
      - postgres_data:/var/lib/postgresql/data

  redis:
    image: redis:7-alpine

  celery:
    build: .
    command: celery -A tasks.celery worker --loglevel=info
    environment:
      - REDIS_URL=redis://redis:6379/0
    depends_on:
      - redis

volumes:
  postgres_data:

3. Gunicorn 生产部署

# 安装
pip install gunicorn

# 启动(4 个 worker 进程)
gunicorn -w 4 -b 0.0.0.0:8000 "run:create_app()"

# 带参数详解
gunicorn \
  -w 4 \                    # worker 数量:通常 CPU 核数 × 2 + 1
  -b 0.0.0.0:8000 \         # 绑定地址
  -k gevent \               # Worker 类型:gevent(异步IO,适合IO密集)
  --max-requests 10000 \    # 每个 worker 处理 10000 请求后重启(防内存泄漏)
  --max-requests-jitter 500 \  # 随机抖动避免同时重启
  --access-logfile - \      # 访问日志输出到 stdout
  --error-logfile - \       # 错误日志输出到 stderr
  --log-level info \
  "run:create_app()"

Worker 类型选择

Worker 类型 适用场景
sync(默认) CPU 密集型、低并发
gevent IO 密集型、高并发(数据库查询、API 调用)
gthread 多线程,共享内存,适合中等并发
uvicorn.workers.UvicornWorker ASGI,FastAPI 部署(Flask 2.0+ 支持 async)

4. 性能优化清单

4.1 数据库层面

# 使用连接池(Flask-SQLAlchemy 默认已具备)
app.config["SQLALCHEMY_ENGINE_OPTIONS"] = {
    "pool_size": 10,           # 连接池大小
    "pool_recycle": 3600,      # 连接回收时间
    "pool_pre_ping": True,     # 使用前检查连接有效性
}

# N+1 查询优化 —— 使用 joinedload
# ❌ 坏:每条 Post 触发一次 User 查询
posts = Post.query.all()
# ✅ 好:一次 JOIN 查询全部加载
from sqlalchemy.orm import joinedload
posts = Post.query.options(joinedload(Post.author)).all()

4.2 缓存策略

# Flask-Caching
from flask_caching import Cache
cache = Cache(config={"CACHE_TYPE": "RedisCache", "CACHE_REDIS_URL": "redis://localhost:6379/2"})

@main_bp.route("/hot-posts")
@cache.cached(timeout=300)  # 缓存 5 分钟
def hot_posts():
    return jsonify(get_expensive_data())

4.3 静态文件优化

# Nginx 配置:让 Nginx 直接服务静态文件,不经过 Flask
location /static/ {
    alias /app/static/;
    expires 30d;          # 设置缓存
    add_header Cache-Control "public, immutable";
}

location / {
    proxy_pass http://127.0.0.1:8000;  # 动态请求代理到 Gunicorn
}

思考题

  1. Celery 的 brokerbackend 各起什么作用?可以用同一个 Redis 实例吗?
  2. 为什么生产环境不推荐用 flask run,而要用 Gunicorn?
  3. Docker Compose 中 depends_on 是否保证服务已就绪?如何实现真正的启动顺序控制?
  4. 什么场景适合用 gevent worker 而非默认 sync worker?

参考资料

  1. [1] Miguel Grinberg. Flask Web Development: Developing Web Applications with Python. 2018.
  2. [2] Pallets Project. Flask 官方文档. 2024.
  3. [3] Miguel Grinberg. The Flask Mega-Tutorial. 2024.
  4. [4] Kunal Relan. Building REST APIs with Flask. 2019.