Flask Web 开发

SunnyFan大约 8 分钟约 2502 字

Flask Web 开发

简介

Flask 是 Python 轻量级 Web 框架，以简洁灵活著称。相比 Django 的全功能方案，Flask 只提供路由和模板引擎，其他功能通过扩展按需引入。适合小型项目、API 服务和微服务开发。

特点

1.轻量灵活 — 核心只含路由和模板
2.蓝图模块 — 应用拆分为模块
3.REST API — 快速构建 API 服务
4.扩展丰富 — 数据库/认证/缓存按需引入

基础用法

Hello Flask

# pip install flask
from flask import Flask, jsonify, request

app = Flask(__name__)

@app.route('/')
def hello():
    return 'Hello, Flask!'

@app.route('/api/health')
def health():
    return jsonify({'status': 'ok', 'version': '1.0.0'})

# 路由参数
@app.route('/api/users/<int:user_id>')
def get_user(user_id):
    return jsonify({'id': user_id, 'name': f'User{user_id}'})

if __name__ == '__main__':
    app.run(debug=True, port=5000)

REST API

from flask import Flask, jsonify, request, abort
from dataclasses import dataclass, field
from typing import List

app = Flask(__name__)

# 内存数据存储
@dataclass
class User:
    id: int
    name: str
    email: str

users: List[User] = []
next_id = 1

# GET /api/users — 获取用户列表
@app.route('/api/users', methods=['GET'])
def list_users():
    page = request.args.get('page', 1, type=int)
    size = request.args.get('size', 10, type=int)
    keyword = request.args.get('keyword', '')

    filtered = [u for u in users if keyword.lower() in u.name.lower()]
    start = (page - 1) * size
    items = filtered[start:start + size]

    return jsonify({
        'items': [{'id': u.id, 'name': u.name, 'email': u.email} for u in items],
        'total': len(filtered),
        'page': page,
        'size': size
    })

# POST /api/users — 创建用户
@app.route('/api/users', methods=['POST'])
def create_user():
    global next_id
    data = request.get_json()

    if not data or 'name' not in data or 'email' not in data:
        return jsonify({'error': 'name and email are required'}), 400

    user = User(id=next_id, name=data['name'], email=data['email'])
    users.append(user)
    next_id += 1

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

# PUT /api/users/<id> — 更新用户
@app.route('/api/users/<int:user_id>', methods=['PUT'])
def update_user(user_id):
    data = request.get_json()
    user = next((u for u in users if u.id == user_id), None)
    if not user:
        abort(404)

    user.name = data.get('name', user.name)
    user.email = data.get('email', user.email)

    return jsonify({'id': user.id, 'name': user.name, 'email': user.email})

# DELETE /api/users/<id> — 删除用户
@app.route('/api/users/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
    global users
    user = next((u for u in users if u.id == user_id), None)
    if not user:
        abort(404)

    users = [u for u in users if u.id != user_id]
    return '', 204

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

蓝图模块化

蓝图组织

# blueprints/product.py
from flask import Blueprint, jsonify, request

product_bp = Blueprint('products', __name__, url_prefix='/api/products')

@product_bp.route('', methods=['GET'])
def list_products():
    return jsonify({'products': []})

@product_bp.route('/<int:product_id>', methods=['GET'])
def get_product(product_id):
    return jsonify({'id': product_id, 'name': 'Product'})

@product_bp.route('', methods=['POST'])
def create_product():
    data = request.get_json()
    return jsonify(data), 201

# blueprints/order.py
order_bp = Blueprint('orders', __name__, url_prefix='/api/orders')

@order_bp.route('', methods=['GET'])
def list_orders():
    return jsonify({'orders': []})

# app.py
from flask import Flask
from blueprints.product import product_bp
from blueprints.order import order_bp

app = Flask(__name__)
app.register_blueprint(product_bp)
app.register_blueprint(order_bp)

中间件与错误处理

请求处理

from flask import Flask, jsonify, request, g
import time
import logging

app = Flask(__name__)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 请求计时中间件
@app.before_request
def before_request():
    g.start_time = time.time()

@app.after_request
def after_request(response):
    elapsed = time.time() - g.start_time
    logger.info(f"{request.method} {request.path} → {response.status_code} ({elapsed:.3f}s)")
    return response

# 全局错误处理
@app.errorhandler(404)
def not_found(e):
    return jsonify({'error': 'Resource not found'}), 404

@app.errorhandler(500)
def internal_error(e):
    return jsonify({'error': 'Internal server error'}), 500

@app.errorhandler(400)
def bad_request(e):
    return jsonify({'error': str(e)}), 400

# 统一响应格式
def api_response(data=None, message='success', code=200):
    return jsonify({
        'code': code,
        'message': message,
        'data': data
    }), code

认证与授权

JWT 认证实现

# pip install flask-jwt-extended
from flask import Flask, jsonify, request
from flask_jwt_extended import (
    JWTManager, create_access_token, create_refresh_token,
    jwt_required, get_jwt_identity, get_jwt
)
from werkzeug.security import generate_password_hash, check_password_hash
from datetime import timedelta, datetime, timezone

app = Flask(__name__)
app.config['JWT_SECRET_KEY'] = 'your-secret-key-change-in-production'
app.config['JWT_ACCESS_TOKEN_EXPIRES'] = timedelta(hours=1)
app.config['JWT_REFRESH_TOKEN_EXPIRES'] = timedelta(days=30)
app.config['JWT_BLACKLIST_ENABLED'] = True
app.config['JWT_BLACKLIST_TOKEN_CHECKS'] = ['access', 'refresh']

jwt = JWTManager(app)

# 模拟用户存储
users_db = {}

# Token 黑名单（生产环境用 Redis）
blacklist = set()

@jwt.token_in_blocklist_loader
def check_if_token_revoked(jwt_header, jwt_payload):
    jti = jwt_payload.get('jti')
    return jti in blacklist

@app.route('/api/auth/register', methods=['POST'])
def register():
    data = request.get_json()
    if not data or 'email' not in data or 'password' not in data:
        return jsonify({'error': 'email and password are required'}), 400

    email = data['email']
    if email in users_db:
        return jsonify({'error': 'email already registered'}), 409

    users_db[email] = {
        'email': email,
        'name': data.get('name', ''),
        'password': generate_password_hash(data['password']),
    }
    return jsonify({'message': 'registered', 'email': email}), 201

@app.route('/api/auth/login', methods=['POST'])
def login():
    data = request.get_json()
    email = data.get('email', '')
    password = data.get('password', '')

    user = users_db.get(email)
    if not user or not check_password_hash(user['password'], password):
        return jsonify({'error': 'invalid credentials'}), 401

    access_token = create_access_token(identity=email)
    refresh_token = create_refresh_token(identity=email)
    return jsonify({
        'access_token': access_token,
        'refresh_token': refresh_token,
        'token_type': 'bearer',
    })

@app.route('/api/auth/refresh', methods=['POST'])
@jwt_required(refresh=True)
def refresh():
    current_user = get_jwt_identity()
    new_access_token = create_access_token(identity=current_user)
    return jsonify({'access_token': new_access_token})

@app.route('/api/auth/logout', methods=['POST'])
@jwt_required(verify_jwt_in_request=False)
@jwt_required()
def logout():
    jti = get_jwt()['jti']
    blacklist.add(jti)
    return jsonify({'message': 'logged out'})

@app.route('/api/me', methods=['GET'])
@jwt_required()
def get_me():
    current_user = get_jwt_identity()
    user = users_db.get(current_user)
    return jsonify({
        'email': user['email'],
        'name': user['name'],
    })

角色权限控制

from functools import wraps

# 角色定义
ROLES = {
    'admin': ['read', 'write', 'delete', 'manage_users'],
    'editor': ['read', 'write'],
    'viewer': ['read'],
}

def role_required(*allowed_roles):
    """角色权限装饰器"""
    def decorator(fn):
        @wraps(fn)
        @jwt_required()
        def wrapper(*args, **kwargs):
            current_user = get_jwt_identity()
            user = users_db.get(current_user)
            user_role = user.get('role', 'viewer') if user else 'viewer'

            if user_role not in allowed_roles:
                return jsonify({'error': 'permission denied'}), 403

            return fn(*args, **kwargs)
        return wrapper
    return decorator

@app.route('/api/admin/users', methods=['GET'])
@role_required('admin')
def admin_list_users():
    return jsonify([
        {'email': u['email'], 'name': u['name'], 'role': u.get('role', 'viewer')}
        for u in users_db.values()
    ])

@app.route('/api/posts', methods=['POST'])
@role_required('admin', 'editor')
def create_post():
    return jsonify({'message': 'post created'}), 201

数据库集成

Flask-SQLAlchemy

# pip install flask-sqlalchemy
from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

db = SQLAlchemy(app)

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(80), nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)

    def to_dict(self):
        return {'id': self.id, 'name': self.name, 'email': self.email}

# 初始化数据库
with app.app_context():
    db.create_all()

@app.route('/api/users', methods=['GET'])
def get_users():
    users = User.query.all()
    return jsonify([u.to_dict() for u in users])

测试与部署

单元测试

import pytest
import json
from app import app, db as _db

@pytest.fixture
def client():
    app.config['TESTING'] = True
    app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///:memory:'

    with app.test_client() as client:
        with app.app_context():
            _db.create_all()
            yield client
            _db.drop_all()

def test_create_user(client):
    response = client.post('/api/users', json={
        'name': '测试用户',
        'email': 'test@example.com',
    })
    assert response.status_code == 201
    data = response.get_json()
    assert data['name'] == '测试用户'
    assert data['email'] == 'test@example.com'

def test_get_users(client):
    # 先创建用户
    client.post('/api/users', json={
        'name': '张三', 'email': 'zhang@example.com'
    })
    response = client.get('/api/users')
    assert response.status_code == 200
    data = response.get_json()
    assert data['total'] == 1

def test_update_user(client):
    # 创建后更新
    client.post('/api/users', json={
        'name': '原名称', 'email': 'update@example.com'
    })
    response = client.put('/api/users/1', json={'name': '新名称'})
    assert response.status_code == 200
    assert response.get_json()['name'] == '新名称'

def test_delete_user(client):
    client.post('/api/users', json={
        'name': '待删除', 'email': 'delete@example.com'
    })
    response = client.delete('/api/users/1')
    assert response.status_code == 204

def test_404_not_found(client):
    response = client.get('/api/users/99999')
    assert response.status_code == 404

Gunicorn 生产部署

# 安装 Gunicorn
pip install gunicorn

# 启动（4 worker，每个 worker 2 线程）
gunicorn -w 4 --threads 2 -b 0.0.0.0:5000 "app:app"

# 带超时和日志
gunicorn -w 4 --timeout 120 --access-logfile - --error-logfile - "app:app"

# 使用配置文件 gunicorn_config.py

# gunicorn_config.py
import multiprocessing

bind = "0.0.0.0:5000"
workers = multiprocessing.cpu_count() * 2 + 1
threads = 2
worker_class = "gthread"
timeout = 120
keepalive = 5

# 日志配置
accesslog = "-"
errorlog = "-"
loglevel = "info"

# 安全配置
limit_request_line = 8190
limit_request_fields = 100

常见问题与最佳实践

应用工厂模式

# app/factory.py
from flask import Flask, jsonify
from flask_sqlalchemy import SQLAlchemy
from flask_cors import CORS

db = SQLAlchemy()

def create_app(config_name='default'):
    app = Flask(__name__)

    # 加载配置
    config_map = {
        'development': 'config.DevelopmentConfig',
        'production': 'config.ProductionConfig',
        'testing': 'config.TestingConfig',
    }
    app.config.from_object(config_map.get(config_name))

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

    # 注册蓝图
    from app.api.v1 import api_v1_bp
    app.register_blueprint(api_v1_bp, url_prefix='/api/v1')

    # 错误处理
    @app.errorhandler(Exception)
    def handle_exception(e):
        return jsonify({'error': str(e)}), 500

    return app

# config.py
class BaseConfig:
    SQLALCHEMY_TRACK_MODIFICATIONS = False

class DevelopmentConfig(BaseConfig):
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = 'sqlite:///dev.db'

class ProductionConfig(BaseConfig):
    DEBUG = False
    SQLALCHEMY_DATABASE_URI = 'postgresql://user:pass@localhost/prod'

class TestingConfig(BaseConfig):
    TESTING = True
    SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:'

# 使用工厂模式
# from app.factory import create_app
# app = create_app('development')

请求限流与安全

# pip install flask-limiter
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

limiter = Limiter(
    app=app,
    key_func=get_remote_address,
    default_limits=["200 per day", "50 per hour"],
)

# 对特定路由设置更严格的限流
@app.route('/api/auth/login', methods=['POST'])
@limiter.limit("5 per minute")
def login():
    return jsonify({'token': 'xxx'})

# 全局限流错误处理
@app.errorhandler(429)
def ratelimit_handler(e):
    return jsonify({
        'error': 'too many requests',
        'retry_after': e.description,
    }), 429

优点

1.轻量简洁 — 核心代码少，容易理解
2.灵活扩展 — 按需引入数据库/认证等
3.开发快速 — 适合小型项目和原型
4.文档完善 — 官方文档清晰

缺点

1.异步支持 — 原生不支持 async/await
2.大型项目 — 不如 Django 全功能
3.ORM 选型 — 需要自己选择和配置
4.生产部署 — 需要 Gunicorn/uWSGI

Flask 核心：路由（@app.route）、请求（request）、响应（jsonify）。REST API 用 GET/POST/PUT/DELETE 方法。蓝图（Blueprint）实现模块化。中间件用 before_request/after_request。数据库集成用 Flask-SQLAlchemy。生产部署用 Gunicorn + Nginx。适合小型 API 服务和微服务，大型项目推荐 Django 或 FastAPI。

关键知识点

先区分这篇内容属于语法能力、工程能力，还是生态工具能力。
Python 的开发效率来自生态，但可维护性来自结构、测试和规范。
脚本一旦进入长期维护，就必须按项目来治理。
框架与语言特性类主题要同时理解运行方式和工程组织方式。

项目落地视角

统一虚拟环境、依赖锁定、格式化和日志方案。
把入口、配置、业务逻辑和工具函数拆开，避免单文件膨胀。
对网络请求、文件读写和数据处理结果做异常与样本校验。
明确项目入口、配置管理、依赖管理、日志和测试策略。

常见误区

把临时脚本直接当生产代码使用。
忽略依赖版本、编码、路径和时区差异。
只会写 happy path，没有补超时、重试和资源释放。
把 notebook 或脚本风格直接带入长期维护项目。

进阶路线

把类型注解、测试、打包和部署纳入统一工程流程。
继续向异步、性能、数据管线和框架源码层深入。
把常用脚本抽成可复用库或 CLI 工具，而不是复制粘贴。
继续补齐部署、打包、监控和性能调优能力。

适用场景

当你准备把《Flask Web 开发》真正落到项目里时，最适合先在一个独立模块或最小样例里验证关键路径。
适合脚本自动化、数据处理、Web 开发和测试工具建设。
当需求强调快速迭代和丰富生态时，Python 往往能快速起步。

落地建议

统一使用虚拟环境与依赖锁定，避免环境漂移。
对核心函数补类型注解、异常处理和日志，减少“脚本黑盒”。
一旦脚本进入生产链路，及时补测试和监控。

排错清单

先确认当前解释器、虚拟环境和依赖版本是否正确。
检查编码、路径、时区和第三方库行为差异。
排查同步阻塞、数据库连接未释放或网络请求无超时。

复盘问题

如果把《Flask Web 开发》放进你的当前项目，最先要验证的输入、输出和失败路径分别是什么？
《Flask Web 开发》最容易在什么规模、什么边界条件下暴露问题？你会用什么指标或日志去确认？
相比默认实现或替代方案，采用《Flask Web 开发》最大的收益和代价分别是什么？

Flask Web 开发

Flask Web 开发

简介

特点

基础用法

Hello Flask

REST API

蓝图模块化

蓝图组织

中间件与错误处理

请求处理

认证与授权

JWT 认证实现

角色权限控制

数据库集成

Flask-SQLAlchemy

测试与部署

单元测试

Gunicorn 生产部署

常见问题与最佳实践

应用工厂模式

请求限流与安全

优点

缺点

总结

关键知识点

项目落地视角

常见误区

进阶路线

适用场景

落地建议

排错清单

复盘问题

延伸阅读