3.2. Flask 实战

Flask 是一个轻量级的 WSGI Web 框架，以简单灵活著称，适合快速开发和小到中型项目。

3.2.1. 快速入门

3.2.1.1. 基础应用

from flask import Flask, request, jsonify

app = Flask(__name__)

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

@app.route('/api/items/<int:item_id>')
def get_item(item_id):
    return jsonify({"id": item_id})

@app.route('/api/items', methods=['POST'])
def create_item():
    data = request.get_json()
    return jsonify(data), 201

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

3.2.1.2. 应用工厂模式

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

db = SQLAlchemy()
migrate = Migrate()

def create_app(config_name='development'):
    app = Flask(__name__)
    
    # 加载配置
    app.config.from_object(config[config_name])
    
    # 初始化扩展
    db.init_app(app)
    migrate.init_app(app, db)
    
    # 注册蓝图
    from app.views import main_bp, api_bp
    app.register_blueprint(main_bp)
    app.register_blueprint(api_bp, url_prefix='/api')
    
    # 错误处理
    register_error_handlers(app)
    
    return app

def register_error_handlers(app):
    @app.errorhandler(404)
    def not_found(error):
        return jsonify({"error": "Not found"}), 404
    
    @app.errorhandler(500)
    def internal_error(error):
        db.session.rollback()
        return jsonify({"error": "Internal server error"}), 500

3.2.2. 蓝图（Blueprint）

# app/views/users.py
from flask import Blueprint, request, jsonify
from app.models import User
from app.services import UserService

users_bp = Blueprint('users', __name__)

@users_bp.route('/', methods=['GET'])
def list_users():
    page = request.args.get('page', 1, type=int)
    per_page = request.args.get('per_page', 20, type=int)
    
    pagination = User.query.paginate(page=page, per_page=per_page)
    
    return jsonify({
        'users': [u.to_dict() for u in pagination.items],
        'total': pagination.total,
        'pages': pagination.pages,
        'current_page': page
    })

@users_bp.route('/<int:user_id>', methods=['GET'])
def get_user(user_id):
    user = User.query.get_or_404(user_id)
    return jsonify(user.to_dict())

@users_bp.route('/', methods=['POST'])
def create_user():
    data = request.get_json()
    
    # 验证
    if not data.get('email'):
        return jsonify({"error": "Email required"}), 400
    
    user = UserService.create_user(data)
    return jsonify(user.to_dict()), 201

3.2.3. 请求处理

3.2.3.1. 获取请求数据

from flask import request

@app.route('/api/submit', methods=['POST'])
def submit():
    # URL 参数: /api/submit?page=1
    page = request.args.get('page', 1, type=int)
    
    # JSON 数据
    json_data = request.get_json()
    
    # 表单数据
    form_data = request.form.get('field_name')
    
    # 文件上传
    file = request.files.get('upload')
    if file:
        file.save(f'/uploads/{file.filename}')
    
    # 请求头
    auth_header = request.headers.get('Authorization')
    
    # Cookies
    session_id = request.cookies.get('session_id')
    
    return jsonify({"status": "received"})

3.2.3.2. 响应处理

from flask import make_response, jsonify, redirect, url_for

@app.route('/api/custom-response')
def custom_response():
    response = make_response(jsonify({"data": "value"}))
    response.headers['X-Custom-Header'] = 'custom-value'
    response.set_cookie('session', 'value', httponly=True, secure=True)
    return response

@app.route('/redirect-example')
def redirect_example():
    return redirect(url_for('get_user', user_id=1))

@app.route('/api/download')
def download():
    data = generate_csv()
    response = make_response(data)
    response.headers['Content-Type'] = 'text/csv'
    response.headers['Content-Disposition'] = 'attachment; filename=data.csv'
    return response

3.2.4. 数据库操作

3.2.4.1. SQLAlchemy 模型

# app/models/user.py
from app import db
from datetime import datetime
from werkzeug.security import generate_password_hash, check_password_hash

class User(db.Model):
    __tablename__ = 'users'
    
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)
    password_hash = db.Column(db.String(128))
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    
    # 关系
    posts = db.relationship('Post', backref='author', lazy='dynamic')
    
    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 to_dict(self):
        return {
            'id': self.id,
            'username': self.username,
            'email': self.email,
            'created_at': self.created_at.isoformat()
        }
    
    def __repr__(self):
        return f'<User {self.username}>'

3.2.4.2. 数据库查询

from app.models import User

# 基本查询
user = User.query.get(1)
user = User.query.filter_by(username='alice').first()
user = User.query.filter(User.email.like('%@example.com')).all()

# 复杂查询
from sqlalchemy import and_, or_

users = User.query.filter(
    and_(
        User.created_at >= start_date,
        or_(
            User.username.contains('admin'),
            User.email.endswith('@company.com')
        )
    )
).order_by(User.created_at.desc()).limit(10).all()

# 分页
pagination = User.query.paginate(page=1, per_page=20)
users = pagination.items
total = pagination.total

# 聚合
from sqlalchemy import func
count = db.session.query(func.count(User.id)).scalar()

3.2.4.3. 事务处理

from app import db

def transfer_funds(from_id, to_id, amount):
    try:
        from_account = Account.query.get(from_id)
        to_account = Account.query.get(to_id)
        
        if from_account.balance < amount:
            raise ValueError("Insufficient funds")
        
        from_account.balance -= amount
        to_account.balance += amount
        
        db.session.commit()
    except Exception as e:
        db.session.rollback()
        raise

3.2.5. 认证与授权

3.2.5.1. Flask-Login 集成

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

login_manager = LoginManager()

class User(UserMixin, db.Model):
    # UserMixin 提供 is_authenticated, is_active, is_anonymous, get_id
    pass

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

@app.route('/login', methods=['POST'])
def login():
    data = request.get_json()
    user = User.query.filter_by(email=data['email']).first()
    
    if user and user.check_password(data['password']):
        login_user(user, remember=data.get('remember', False))
        return jsonify({"message": "Logged in"})
    
    return jsonify({"error": "Invalid credentials"}), 401

@app.route('/logout')
@login_required
def logout():
    logout_user()
    return jsonify({"message": "Logged out"})

@app.route('/profile')
@login_required
def profile():
    return jsonify(current_user.to_dict())

3.2.5.2. JWT 认证

from flask import Flask, request, jsonify
from functools import wraps
import jwt
from datetime import datetime, timedelta

app = Flask(__name__)
app.config['SECRET_KEY'] = 'your-secret-key'

def token_required(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        token = request.headers.get('Authorization', '').replace('Bearer ', '')
        
        if not token:
            return jsonify({'error': 'Token missing'}), 401
        
        try:
            data = jwt.decode(token, app.config['SECRET_KEY'], algorithms=['HS256'])
            current_user = User.query.get(data['user_id'])
        except jwt.ExpiredSignatureError:
            return jsonify({'error': 'Token expired'}), 401
        except jwt.InvalidTokenError:
            return jsonify({'error': 'Invalid token'}), 401
        
        return f(current_user, *args, **kwargs)
    return decorated

@app.route('/api/login', methods=['POST'])
def api_login():
    data = request.get_json()
    user = User.query.filter_by(email=data['email']).first()
    
    if user and user.check_password(data['password']):
        token = jwt.encode({
            'user_id': user.id,
            'exp': datetime.utcnow() + timedelta(hours=24)
        }, app.config['SECRET_KEY'])
        
        return jsonify({'token': token})
    
    return jsonify({'error': 'Invalid credentials'}), 401

@app.route('/api/protected')
@token_required
def protected(current_user):
    return jsonify({'user': current_user.to_dict()})

3.2.6. 中间件和钩子

from flask import g, request
import time

# 请求前
@app.before_request
def before_request():
    g.start_time = time.time()
    g.user = None
    
    token = request.headers.get('Authorization')
    if token:
        g.user = verify_token(token)

# 请求后
@app.after_request
def after_request(response):
    # 添加响应头
    response.headers['X-Process-Time'] = str(time.time() - g.start_time)
    return response

# 请求结束（无论成功失败）
@app.teardown_request
def teardown_request(exception=None):
    db = getattr(g, '_database', None)
    if db is not None:
        db.close()

# 第一次请求前
@app.before_first_request
def before_first_request():
    # 初始化操作
    pass

3.2.7. 测试

# tests/test_api.py
import pytest
from app import create_app, db
from app.models import User

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

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

@pytest.fixture
def auth_header(client):
    # 创建测试用户
    response = client.post('/api/register', json={
        'email': 'test@example.com',
        'password': 'password123'
    })
    
    response = client.post('/api/login', json={
        'email': 'test@example.com',
        'password': 'password123'
    })
    token = response.get_json()['token']
    
    return {'Authorization': f'Bearer {token}'}

def test_get_users(client, auth_header):
    response = client.get('/api/users', headers=auth_header)
    assert response.status_code == 200

def test_create_user(client):
    response = client.post('/api/users', json={
        'email': 'new@example.com',
        'username': 'newuser'
    })
    assert response.status_code == 201
    assert response.get_json()['email'] == 'new@example.com'

3.2.8. 部署

3.2.8.1. Gunicorn

# 安装
pip install gunicorn

# 运行
gunicorn -w 4 -b 0.0.0.0:8000 "app:create_app()"

# 生产配置
gunicorn \
    --workers 4 \
    --threads 2 \
    --worker-class gthread \
    --bind 0.0.0.0:8000 \
    --access-logfile - \
    --error-logfile - \
    "app:create_app('production')"

3.2.8.2. Docker

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["gunicorn", "-w", "4", "-b", "0.0.0.0:8000", "app:create_app()"]

3.2.9. 最佳实践

项目组织

使用应用工厂：便于测试和配置
使用蓝图：模块化路由
分离配置：不同环境不同配置
使用服务层：业务逻辑与视图分离

安全实践

永远不要信任用户输入
使用 CSRF 保护（flask-wtf）
设置安全的 Cookie（httponly, secure）
使用环境变量存储密钥
定期更新依赖

性能优化

使用数据库索引
实现分页
使用缓存（Flask-Caching）
异步任务（Celery）
连接池