Back to Details

flask

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

Flask - Lightweight Python Web Framework

Flask - 轻量级Python Web框架

Overview

概述

Flask is a micro-framework for Python web development, designed for building microservices, REST APIs, and flexible web applications. Its minimalist core and extensive extension ecosystem make it ideal for projects requiring lightweight architecture, rapid development, and full control over components.
Key Features:
  • Micro-framework philosophy (minimal core, extensible)
  • Flask-RESTful for API development
  • Blueprints for modular application structure
  • SQLAlchemy integration via Flask-SQLAlchemy
  • Jinja2 templating engine
  • Built-in development server with auto-reload
  • Werkzeug WSGI toolkit foundation
  • Large extension ecosystem (Flask-Login, Flask-JWT, Flask-CORS)
  • Production deployment with Gunicorn/uWSGI
Installation:
bash
undefined
Flask是一款用于Python Web开发的微框架，专为构建微服务、REST API和灵活的Web应用而设计。其极简的核心和丰富的扩展生态系统，使其成为需要轻量级架构、快速开发以及对组件完全控制的项目的理想选择。
核心特性:
  • 微框架理念（极简核心，可扩展）
  • 用于API开发的Flask-RESTful
  • 用于模块化应用结构的Blueprints
  • 通过Flask-SQLAlchemy集成SQLAlchemy
  • Jinja2模板引擎
  • 自带支持自动重载的开发服务器
  • 基于Werkzeug WSGI工具包
  • 庞大的扩展生态系统（Flask-Login、Flask-JWT、Flask-CORS等）
  • 可通过Gunicorn/uWSGI部署到生产环境
安装:
bash
undefined

Basic Flask

基础Flask

pip install flask
pip install flask

Flask with common extensions

包含常用扩展的Flask

pip install flask flask-restful flask-sqlalchemy flask-login flask-cors
pip install flask flask-restful flask-sqlalchemy flask-login flask-cors

With database support

带数据库支持

pip install flask flask-sqlalchemy psycopg2-binary # PostgreSQL
pip install flask flask-sqlalchemy psycopg2-binary # PostgreSQL

Full microservices stack

完整微服务栈

pip install flask flask-restful marshmallow flask-jwt-extended redis
undefined
pip install flask flask-restful marshmallow flask-jwt-extended redis
undefined

Basic Application Patterns

基础应用模式

1. Minimal Flask App

1. 最简Flask应用

python
undefined
python
undefined

app.py

app.py

from flask import Flask, jsonify, request
app = Flask(name)
@app.route('/') def hello(): return jsonify({"message": "Hello, World!"})
@app.route('/api/users/int:user_id') def get_user(user_id): return jsonify({"id": user_id, "name": f"User {user_id}"})
@app.route('/api/users', methods=['POST']) def create_user(): data = request.get_json() return jsonify({"id": 123, **data}), 201
if name == 'main': app.run(debug=True, host='0.0.0.0', port=5000)

**Run:**
```bash
from flask import Flask, jsonify, request
app = Flask(name)
@app.route('/') def hello(): return jsonify({"message": "Hello, World!"})
@app.route('/api/users/int:user_id') def get_user(user_id): return jsonify({"id": user_id, "name": f"User {user_id}"})
@app.route('/api/users', methods=['POST']) def create_user(): data = request.get_json() return jsonify({"id": 123, **data}), 201
if name == 'main': app.run(debug=True, host='0.0.0.0', port=5000)

**运行**:
```bash

Development server

开发服务器

python app.py
python app.py

Or using flask CLI

或使用Flask CLI

export FLASK_APP=app.py export FLASK_ENV=development flask run
export FLASK_APP=app.py export FLASK_ENV=development flask run

Custom port

自定义端口

flask run --port 8000 --host 0.0.0.0
undefined
flask run --port 8000 --host 0.0.0.0
undefined

2. Application Factory Pattern

2. 应用工厂模式

Recommended for production and testing:
python
undefined
推荐用于生产和测试环境:
python
undefined

app/init.py

app/init.py

from flask import Flask from app.config import Config from app.extensions import db, migrate, jwt
def create_app(config_class=Config): """Application factory pattern.""" app = Flask(name) app.config.from_object(config_class)
# Initialize extensions
db.init_app(app)
migrate.init_app(app, db)
jwt.init_app(app)

# Register blueprints
from app.routes import api_bp, auth_bp
app.register_blueprint(api_bp, url_prefix='/api')
app.register_blueprint(auth_bp, url_prefix='/auth')

return app
from flask import Flask from app.config import Config from app.extensions import db, migrate, jwt
def create_app(config_class=Config): """应用工厂模式。""" app = Flask(name) app.config.from_object(config_class)
# 初始化扩展
db.init_app(app)
migrate.init_app(app, db)
jwt.init_app(app)

# 注册蓝图
from app.routes import api_bp, auth_bp
app.register_blueprint(api_bp, url_prefix='/api')
app.register_blueprint(auth_bp, url_prefix='/auth')

return app

app/extensions.py

app/extensions.py

from flask_sqlalchemy import SQLAlchemy from flask_migrate import Migrate from flask_jwt_extended import JWTManager
db = SQLAlchemy() migrate = Migrate() jwt = JWTManager()
from flask_sqlalchemy import SQLAlchemy from flask_migrate import Migrate from flask_jwt_extended import JWTManager
db = SQLAlchemy() migrate = Migrate() jwt = JWTManager()

app/config.py

app/config.py

import os
class Config: SECRET_KEY = os.environ.get('SECRET_KEY') or 'dev-secret-key' SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or 'sqlite:///app.db' SQLALCHEMY_TRACK_MODIFICATIONS = False JWT_SECRET_KEY = os.environ.get('JWT_SECRET_KEY') or 'jwt-secret'
class DevelopmentConfig(Config): DEBUG = True TESTING = False
class ProductionConfig(Config): DEBUG = False TESTING = False
import os
class Config: SECRET_KEY = os.environ.get('SECRET_KEY') or 'dev-secret-key' SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or 'sqlite:///app.db' SQLALCHEMY_TRACK_MODIFICATIONS = False JWT_SECRET_KEY = os.environ.get('JWT_SECRET_KEY') or 'jwt-secret'
class DevelopmentConfig(Config): DEBUG = True TESTING = False
class ProductionConfig(Config): DEBUG = False TESTING = False

run.py

run.py

from app import create_app
app = create_app()
if name == 'main': app.run()

**Run:**
```bash
export FLASK_APP=run.py
flask run
from app import create_app
app = create_app()
if name == 'main': app.run()

**运行**:
```bash
export FLASK_APP=run.py
flask run

3. Request and Response Handling

3. 请求与响应处理

python
from flask import Flask, request, jsonify, make_response, abort

app = Flask(__name__)

@app.route('/api/data', methods=['GET', 'POST'])
def handle_data():
    # GET request
    if request.method == 'GET':
        # Query parameters
        page = request.args.get('page', 1, type=int)
        limit = request.args.get('limit', 10, type=int)

        return jsonify({
            "page": page,
            "limit": limit,
            "data": [...]
        })

    # POST request
    if request.method == 'POST':
        # JSON body
        data = request.get_json()

        # Validation
        if not data or 'name' not in data:
            abort(400, description="Missing required field: name")

        # Custom response with headers
        response = make_response(jsonify({"id": 1, **data}), 201)
        response.headers['X-Custom-Header'] = 'value'
        return response
python
from flask import Flask, request, jsonify, make_response, abort

app = Flask(__name__)

@app.route('/api/data', methods=['GET', 'POST'])
def handle_data():
    # GET请求
    if request.method == 'GET':
        # 查询参数
        page = request.args.get('page', 1, type=int)
        limit = request.args.get('limit', 10, type=int)

        return jsonify({
            "page": page,
            "limit": limit,
            "data": [...]
        })

    # POST请求
    if request.method == 'POST':
        # JSON请求体
        data = request.get_json()

        # 验证
        if not data or 'name' not in data:
            abort(400, description="缺少必填字段: name")

        # 带自定义头的响应
        response = make_response(jsonify({"id": 1, **data}), 201)
        response.headers['X-Custom-Header'] = 'value'
        return response

Error handling

错误处理

@app.errorhandler(404) def not_found(error): return jsonify({"error": "Resource not found"}), 404
@app.errorhandler(400) def bad_request(error): return jsonify({"error": str(error.description)}), 400
undefined
@app.errorhandler(404) def not_found(error): return jsonify({"error": "资源未找到"}), 404
@app.errorhandler(400) def bad_request(error): return jsonify({"error": str(error.description)}), 400
undefined

Blueprints - Modular Applications

Blueprints - 模块化应用

Blueprint Structure

蓝图结构

app/
├── __init__.py          # Application factory
├── extensions.py        # Extension instances
├── config.py            # Configuration
├── models/
│   ├── __init__.py
│   ├── user.py
│   └── product.py
├── routes/
│   ├── __init__.py
│   ├── auth.py          # Authentication routes
│   ├── users.py         # User management routes
│   └── products.py      # Product routes
└── services/
    ├── __init__.py
    ├── user_service.py
    └── auth_service.py
app/
├── __init__.py          # 应用工厂
├── extensions.py        # 扩展实例
├── config.py            # 配置文件
├── models/
│   ├── __init__.py
│   ├── user.py
│   └── product.py
├── routes/
│   ├── __init__.py
│   ├── auth.py          # 认证路由
│   ├── users.py         # 用户管理路由
│   └── products.py      # 产品路由
└── services/
    ├── __init__.py
    ├── user_service.py
    └── auth_service.py

Blueprint Implementation

蓝图实现

python
undefined
python
undefined

app/routes/users.py

app/routes/users.py

from flask import Blueprint, jsonify, request from app.models.user import User from app.extensions import db
users_bp = Blueprint('users', name)
@users_bp.route('/', methods=['GET']) def list_users(): """List all users.""" page = request.args.get('page', 1, type=int) per_page = request.args.get('per_page', 20, type=int)
users = User.query.paginate(page=page, per_page=per_page)

return jsonify({
    "users": [u.to_dict() for u in users.items],
    "total": users.total,
    "page": users.page,
    "pages": users.pages
})
@users_bp.route('/int:user_id', methods=['GET']) def get_user(user_id): """Get user by ID.""" user = User.query.get_or_404(user_id) return jsonify(user.to_dict())
@users_bp.route('/', methods=['POST']) def create_user(): """Create new user.""" data = request.get_json()
user = User(
    email=data['email'],
    name=data['name']
)
user.set_password(data['password'])

db.session.add(user)
db.session.commit()

return jsonify(user.to_dict()), 201
@users_bp.route('/int:user_id', methods=['PUT']) def update_user(user_id): """Update user.""" user = User.query.get_or_404(user_id) data = request.get_json()
if 'name' in data:
    user.name = data['name']
if 'email' in data:
    user.email = data['email']

db.session.commit()
return jsonify(user.to_dict())
@users_bp.route('/int:user_id', methods=['DELETE']) def delete_user(user_id): """Delete user.""" user = User.query.get_or_404(user_id) db.session.delete(user) db.session.commit()
return '', 204
from flask import Blueprint, jsonify, request from app.models.user import User from app.extensions import db
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)
users = User.query.paginate(page=page, per_page=per_page)

return jsonify({
    "users": [u.to_dict() for u in users.items],
    "total": users.total,
    "page": users.page,
    "pages": users.pages
})
@users_bp.route('/int:user_id', methods=['GET']) def get_user(user_id): """根据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()
user = User(
    email=data['email'],
    name=data['name']
)
user.set_password(data['password'])

db.session.add(user)
db.session.commit()

return jsonify(user.to_dict()), 201
@users_bp.route('/int:user_id', methods=['PUT']) def update_user(user_id): """更新用户信息。""" user = User.query.get_or_404(user_id) data = request.get_json()
if 'name' in data:
    user.name = data['name']
if 'email' in data:
    user.email = data['email']

db.session.commit()
return jsonify(user.to_dict())
@users_bp.route('/int:user_id', methods=['DELETE']) def delete_user(user_id): """删除用户。""" user = User.query.get_or_404(user_id) db.session.delete(user) db.session.commit()
return '', 204

app/init.py

app/init.py

def create_app(): app = Flask(name)
# Register blueprints
from app.routes.users import users_bp
from app.routes.products import products_bp

app.register_blueprint(users_bp, url_prefix='/api/users')
app.register_blueprint(products_bp, url_prefix='/api/products')

return app
undefined
def create_app(): app = Flask(name)
# 注册蓝图
from app.routes.users import users_bp
from app.routes.products import products_bp

app.register_blueprint(users_bp, url_prefix='/api/users')
app.register_blueprint(products_bp, url_prefix='/api/products')

return app
undefined

Flask-RESTful for APIs

Flask-RESTful 用于API开发

Basic REST API

基础REST API

python
undefined
python
undefined

app/api/resources.py

app/api/resources.py

from flask import request from flask_restful import Resource, Api, reqparse, fields, marshal_with from app.models.user import User from app.extensions import db
from flask import request from flask_restful import Resource, Api, reqparse, fields, marshal_with from app.models.user import User from app.extensions import db

Response serialization

响应序列化

user_fields = { 'id': fields.Integer, 'email': fields.String, 'name': fields.String, 'created_at': fields.DateTime(dt_format='iso8601') }
class UserListResource(Resource): """User collection endpoint."""
@marshal_with(user_fields)
def get(self):
    """List all users."""
    users = User.query.all()
    return users

def post(self):
    """Create new user."""
    parser = reqparse.RequestParser()
    parser.add_argument('email', required=True, help='Email is required')
    parser.add_argument('name', required=True, help='Name is required')
    parser.add_argument('password', required=True, help='Password is required')
    args = parser.parse_args()

    user = User(email=args['email'], name=args['name'])
    user.set_password(args['password'])

    db.session.add(user)
    db.session.commit()

    return {'id': user.id, 'email': user.email, 'name': user.name}, 201
class UserResource(Resource): """Single user endpoint."""
@marshal_with(user_fields)
def get(self, user_id):
    """Get user by ID."""
    user = User.query.get_or_404(user_id)
    return user

@marshal_with(user_fields)
def put(self, user_id):
    """Update user."""
    user = User.query.get_or_404(user_id)

    parser = reqparse.RequestParser()
    parser.add_argument('name')
    parser.add_argument('email')
    args = parser.parse_args()

    if args['name']:
        user.name = args['name']
    if args['email']:
        user.email = args['email']

    db.session.commit()
    return user

def delete(self, user_id):
    """Delete user."""
    user = User.query.get_or_404(user_id)
    db.session.delete(user)
    db.session.commit()

    return '', 204
user_fields = { 'id': fields.Integer, 'email': fields.String, 'name': fields.String, 'created_at': fields.DateTime(dt_format='iso8601') }
class UserListResource(Resource): """用户集合端点。"""
@marshal_with(user_fields)
def get(self):
    """列出所有用户。"""
    users = User.query.all()
    return users

def post(self):
    """创建新用户。"""
    parser = reqparse.RequestParser()
    parser.add_argument('email', required=True, help='Email是必填项')
    parser.add_argument('name', required=True, help='Name是必填项')
    parser.add_argument('password', required=True, help='Password是必填项')
    args = parser.parse_args()

    user = User(email=args['email'], name=args['name'])
    user.set_password(args['password'])

    db.session.add(user)
    db.session.commit()

    return {'id': user.id, 'email': user.email, 'name': user.name}, 201
class UserResource(Resource): """单个用户端点。"""
@marshal_with(user_fields)
def get(self, user_id):
    """根据ID获取用户。"""
    user = User.query.get_or_404(user_id)
    return user

@marshal_with(user_fields)
def put(self, user_id):
    """更新用户信息。"""
    user = User.query.get_or_404(user_id)

    parser = reqparse.RequestParser()
    parser.add_argument('name')
    parser.add_argument('email')
    args = parser.parse_args()

    if args['name']:
        user.name = args['name']
    if args['email']:
        user.email = args['email']

    db.session.commit()
    return user

def delete(self, user_id):
    """删除用户。"""
    user = User.query.get_or_404(user_id)
    db.session.delete(user)
    db.session.commit()

    return '', 204

app/init.py

app/init.py

from flask_restful import Api
def create_app(): app = Flask(name) api = Api(app, prefix='/api/v1')
# Register resources
from app.api.resources import UserListResource, UserResource

api.add_resource(UserListResource, '/users')
api.add_resource(UserResource, '/users/<int:user_id>')

return app
undefined
from flask_restful import Api
def create_app(): app = Flask(name) api = Api(app, prefix='/api/v1')
# 注册资源
from app.api.resources import UserListResource, UserResource

api.add_resource(UserListResource, '/users')
api.add_resource(UserResource, '/users/<int:user_id>')

return app
undefined

Request Validation

请求验证

Marshmallow Schemas

Marshmallow 模式

python
undefined
python
undefined

app/schemas/user_schema.py

app/schemas/user_schema.py

from marshmallow import Schema, fields, validate, validates, ValidationError
class UserSchema(Schema): """User validation schema."""
id = fields.Int(dump_only=True)
email = fields.Email(required=True, validate=validate.Length(max=120))
name = fields.Str(required=True, validate=validate.Length(min=2, max=80))
password = fields.Str(
    required=True,
    load_only=True,
    validate=validate.Length(min=8)
)
created_at = fields.DateTime(dump_only=True)

@validates('email')
def validate_email(self, value):
    """Check email uniqueness."""
    from app.models.user import User
    if User.query.filter_by(email=value).first():
        raise ValidationError('Email already registered')
class UserUpdateSchema(Schema): """User update schema (partial updates)."""
email = fields.Email(validate=validate.Length(max=120))
name = fields.Str(validate=validate.Length(min=2, max=80))
password = fields.Str(load_only=True, validate=validate.Length(min=8))
from marshmallow import Schema, fields, validate, validates, ValidationError
class UserSchema(Schema): """用户验证模式。"""
id = fields.Int(dump_only=True)
email = fields.Email(required=True, validate=validate.Length(max=120))
name = fields.Str(required=True, validate=validate.Length(min=2, max=80))
password = fields.Str(
    required=True,
    load_only=True,
    validate=validate.Length(min=8)
)
created_at = fields.DateTime(dump_only=True)

@validates('email')
def validate_email(self, value):
    """检查邮箱唯一性。"""
    from app.models.user import User
    if User.query.filter_by(email=value).first():
        raise ValidationError('该邮箱已被注册')
class UserUpdateSchema(Schema): """用户更新模式（支持部分更新）。"""
email = fields.Email(validate=validate.Length(max=120))
name = fields.Str(validate=validate.Length(min=2, max=80))
password = fields.Str(load_only=True, validate=validate.Length(min=8))

Usage in routes

在路由中使用

from marshmallow import ValidationError from app.schemas.user_schema import UserSchema, UserUpdateSchema
user_schema = UserSchema() users_schema = UserSchema(many=True) user_update_schema = UserUpdateSchema()
@users_bp.route('/', methods=['POST']) def create_user(): """Create user with validation.""" try: # Validate and deserialize data = user_schema.load(request.get_json()) except ValidationError as err: return jsonify({"errors": err.messages}), 400
user = User(**data)
db.session.add(user)
db.session.commit()

# Serialize response
return user_schema.dump(user), 201
@users_bp.route('/', methods=['GET']) def list_users(): """List users with serialization.""" users = User.query.all() return jsonify(users_schema.dump(users))
undefined
from marshmallow import ValidationError from app.schemas.user_schema import UserSchema, UserUpdateSchema
user_schema = UserSchema() users_schema = UserSchema(many=True) user_update_schema = UserUpdateSchema()
@users_bp.route('/', methods=['POST']) def create_user(): """通过验证创建用户。""" try: # 验证并反序列化 data = user_schema.load(request.get_json()) except ValidationError as err: return jsonify({"errors": err.messages}), 400
user = User(**data)
db.session.add(user)
db.session.commit()

# 序列化响应
return user_schema.dump(user), 201
@users_bp.route('/', methods=['GET']) def list_users(): """序列化并列出用户。""" users = User.query.all() return jsonify(users_schema.dump(users))
undefined

Pydantic Validation

Pydantic 验证

python
undefined
python
undefined

app/schemas/user_schema_pydantic.py

app/schemas/user_schema_pydantic.py

from pydantic import BaseModel, EmailStr, Field, validator from typing import Optional from datetime import datetime
class UserCreate(BaseModel): """User creation schema."""
email: EmailStr
name: str = Field(..., min_length=2, max_length=80)
password: str = Field(..., min_length=8)

@validator('password')
def password_complexity(cls, v):
    """Validate password complexity."""
    if not any(char.isdigit() for char in v):
        raise ValueError('Password must contain at least one digit')
    if not any(char.isupper() for char in v):
        raise ValueError('Password must contain at least one uppercase letter')
    return v
class UserUpdate(BaseModel): """User update schema."""
email: Optional[EmailStr] = None
name: Optional[str] = Field(None, min_length=2, max_length=80)
class UserResponse(BaseModel): """User response schema."""
id: int
email: str
name: str
created_at: datetime

class Config:
    orm_mode = True
from pydantic import BaseModel, EmailStr, Field, validator from typing import Optional from datetime import datetime
class UserCreate(BaseModel): """用户创建模式。"""
email: EmailStr
name: str = Field(..., min_length=2, max_length=80)
password: str = Field(..., min_length=8)

@validator('password')
def password_complexity(cls, v):
    """验证密码复杂度。"""
    if not any(char.isdigit() for char in v):
        raise ValueError('密码必须包含至少一个数字')
    if not any(char.isupper() for char in v):
        raise ValueError('密码必须包含至少一个大写字母')
    return v
class UserUpdate(BaseModel): """用户更新模式。"""
email: Optional[EmailStr] = None
name: Optional[str] = Field(None, min_length=2, max_length=80)
class UserResponse(BaseModel): """用户响应模式。"""
id: int
email: str
name: str
created_at: datetime

class Config:
    orm_mode = True

Usage

使用示例

from pydantic import ValidationError
@users_bp.route('/', methods=['POST']) def create_user(): """Create user with Pydantic validation.""" try: user_data = UserCreate(**request.get_json()) except ValidationError as e: return jsonify({"errors": e.errors()}), 400
user = User(
    email=user_data.email,
    name=user_data.name
)
user.set_password(user_data.password)

db.session.add(user)
db.session.commit()

return UserResponse.from_orm(user).dict(), 201
undefined
from pydantic import ValidationError
@users_bp.route('/', methods=['POST']) def create_user(): """通过Pydantic验证创建用户。""" try: user_data = UserCreate(**request.get_json()) except ValidationError as e: return jsonify({"errors": e.errors()}), 400
user = User(
    email=user_data.email,
    name=user_data.name
)
user.set_password(user_data.password)

db.session.add(user)
db.session.commit()

return UserResponse.from_orm(user).dict(), 201
undefined

SQLAlchemy Integration

SQLAlchemy 集成

Models with Flask-SQLAlchemy

基于Flask-SQLAlchemy的模型

python
undefined
python
undefined

app/models/user.py

app/models/user.py

from datetime import datetime from werkzeug.security import generate_password_hash, check_password_hash from app.extensions import db
class User(db.Model): """User model."""
__tablename__ = 'users'

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

# Relationships
posts = db.relationship('Post', backref='author', lazy='dynamic', cascade='all, delete-orphan')

def set_password(self, password):
    """Hash and set password."""
    self.password_hash = generate_password_hash(password)

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

def to_dict(self):
    """Serialize to dictionary."""
    return {
        'id': self.id,
        'email': self.email,
        'name': self.name,
        'is_active': self.is_active,
        'created_at': self.created_at.isoformat(),
        'updated_at': self.updated_at.isoformat()
    }

def __repr__(self):
    return f'<User {self.email}>'
from datetime import datetime from werkzeug.security import generate_password_hash, check_password_hash from app.extensions import db
class User(db.Model): """用户模型。"""
__tablename__ = 'users'

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

# 关联关系
posts = db.relationship('Post', backref='author', lazy='dynamic', cascade='all, delete-orphan')

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,
        'email': self.email,
        'name': self.name,
        'is_active': self.is_active,
        'created_at': self.created_at.isoformat(),
        'updated_at': self.updated_at.isoformat()
    }

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

app/models/post.py

app/models/post.py

class Post(db.Model): """Post model."""
__tablename__ = 'posts'

id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(200), nullable=False)
content = db.Column(db.Text, nullable=False)
user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
created_at = db.Column(db.DateTime, default=datetime.utcnow)

def to_dict(self):
    return {
        'id': self.id,
        'title': self.title,
        'content': self.content,
        'user_id': self.user_id,
        'author': self.author.name,
        'created_at': self.created_at.isoformat()
    }
class Post(db.Model): """文章模型。"""
__tablename__ = 'posts'

id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(200), nullable=False)
content = db.Column(db.Text, nullable=False)
user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
created_at = db.Column(db.DateTime, default=datetime.utcnow)

def to_dict(self):
    return {
        'id': self.id,
        'title': self.title,
        'content': self.content,
        'user_id': self.user_id,
        'author': self.author.name,
        'created_at': self.created_at.isoformat()
    }

Database operations

数据库操作示例

@users_bp.route('/int:user_id/posts', methods=['GET']) def get_user_posts(user_id): """Get user posts with pagination.""" user = User.query.get_or_404(user_id) page = request.args.get('page', 1, type=int)
posts = user.posts.paginate(page=page, per_page=20)

return jsonify({
    'posts': [p.to_dict() for p in posts.items],
    'total': posts.total,
    'page': posts.page
})
undefined
@users_bp.route('/int:user_id/posts', methods=['GET']) def get_user_posts(user_id): """分页获取用户的文章。""" user = User.query.get_or_404(user_id) page = request.args.get('page', 1, type=int)
posts = user.posts.paginate(page=page, per_page=20)

return jsonify({
    'posts': [p.to_dict() for p in posts.items],
    'total': posts.total,
    'page': posts.page
})
undefined

Database Migrations

数据库迁移

bash
undefined
bash
undefined

Initialize migrations

初始化迁移

flask db init
flask db init

Create migration

创建迁移脚本

flask db migrate -m "Create users table"
flask db migrate -m "Create users table"

Apply migration

应用迁移

flask db upgrade
flask db upgrade

Rollback

回滚迁移

flask db downgrade

```python
flask db downgrade

```python

migrations/versions/xxx_create_users.py (auto-generated)

migrations/versions/xxx_create_users.py（自动生成）

def upgrade(): op.create_table( 'users', sa.Column('id', sa.Integer(), nullable=False), sa.Column('email', sa.String(length=120), nullable=False), sa.Column('name', sa.String(length=80), nullable=False), sa.Column('password_hash', sa.String(length=200), nullable=False), sa.Column('created_at', sa.DateTime(), nullable=True), sa.PrimaryKeyConstraint('id'), sa.UniqueConstraint('email') ) op.create_index('ix_users_email', 'users', ['email'])
def downgrade(): op.drop_index('ix_users_email', table_name='users') op.drop_table('users')
undefined
def upgrade(): op.create_table( 'users', sa.Column('id', sa.Integer(), nullable=False), sa.Column('email', sa.String(length=120), nullable=False), sa.Column('name', sa.String(length=80), nullable=False), sa.Column('password_hash', sa.String(length=200), nullable=False), sa.Column('created_at', sa.DateTime(), nullable=True), sa.PrimaryKeyConstraint('id'), sa.UniqueConstraint('email') ) op.create_index('ix_users_email', 'users', ['email'])
def downgrade(): op.drop_index('ix_users_email', table_name='users') op.drop_table('users')
undefined

Authentication

认证

Flask-Login Session-Based Auth

基于Flask-Login的会话认证

python
undefined
python
undefined

app/extensions.py

app/extensions.py

from flask_login import LoginManager
login_manager = LoginManager()
from flask_login import LoginManager
login_manager = LoginManager()

app/init.py

app/init.py

def create_app(): app = Flask(name) login_manager.init_app(app) login_manager.login_view = 'auth.login'
@login_manager.user_loader
def load_user(user_id):
    from app.models.user import User
    return User.query.get(int(user_id))

return app
def create_app(): app = Flask(name) login_manager.init_app(app) login_manager.login_view = 'auth.login'
@login_manager.user_loader
def load_user(user_id):
    from app.models.user import User
    return User.query.get(int(user_id))

return app

app/models/user.py

app/models/user.py

from flask_login import UserMixin
class User(UserMixin, db.Model): # ... existing fields ...
def get_id(self):
    return str(self.id)
from flask_login import UserMixin
class User(UserMixin, db.Model): # ... 现有字段 ...
def get_id(self):
    return str(self.id)

app/routes/auth.py

app/routes/auth.py

from flask import Blueprint, request, jsonify from flask_login import login_user, logout_user, login_required, current_user from app.models.user import User
auth_bp = Blueprint('auth', name)
@auth_bp.route('/login', methods=['POST']) def login(): """User login.""" data = request.get_json()
user = User.query.filter_by(email=data['email']).first()
if not user or not user.check_password(data['password']):
    return jsonify({"error": "Invalid credentials"}), 401

login_user(user, remember=data.get('remember', False))
return jsonify(user.to_dict())
@auth_bp.route('/logout', methods=['POST']) @login_required def logout(): """User logout.""" logout_user() return jsonify({"message": "Logged out successfully"})
@auth_bp.route('/me', methods=['GET']) @login_required def get_current_user(): """Get current authenticated user.""" return jsonify(current_user.to_dict())
undefined
from flask import Blueprint, request, jsonify from flask_login import login_user, logout_user, login_required, current_user from app.models.user import User
auth_bp = Blueprint('auth', name)
@auth_bp.route('/login', methods=['POST']) def login(): """用户登录。""" data = request.get_json()
user = User.query.filter_by(email=data['email']).first()
if not user or not user.check_password(data['password']):
    return jsonify({"error": "无效的凭据"}), 401

login_user(user, remember=data.get('remember', False))
return jsonify(user.to_dict())
@auth_bp.route('/logout', methods=['POST']) @login_required def logout(): """用户登出。""" logout_user() return jsonify({"message": "成功登出"})
@auth_bp.route('/me', methods=['GET']) @login_required def get_current_user(): """获取当前已认证用户。""" return jsonify(current_user.to_dict())
undefined

JWT Authentication

JWT 认证

python
undefined
python
undefined

app/extensions.py

app/extensions.py

from flask_jwt_extended import JWTManager
jwt = JWTManager()
from flask_jwt_extended import JWTManager
jwt = JWTManager()

app/init.py

app/init.py

def create_app(): app = Flask(name) app.config['JWT_SECRET_KEY'] = 'super-secret-key' app.config['JWT_ACCESS_TOKEN_EXPIRES'] = timedelta(hours=1) app.config['JWT_REFRESH_TOKEN_EXPIRES'] = timedelta(days=30)
jwt.init_app(app)

return app
def create_app(): app = Flask(name) app.config['JWT_SECRET_KEY'] = 'super-secret-key' app.config['JWT_ACCESS_TOKEN_EXPIRES'] = timedelta(hours=1) app.config['JWT_REFRESH_TOKEN_EXPIRES'] = timedelta(days=30)
jwt.init_app(app)

return app

app/routes/auth.py

app/routes/auth.py

from flask_jwt_extended import ( create_access_token, create_refresh_token, jwt_required, get_jwt_identity )
@auth_bp.route('/login', methods=['POST']) def login(): """Login and return JWT tokens.""" data = request.get_json()
user = User.query.filter_by(email=data['email']).first()
if not user or not user.check_password(data['password']):
    return jsonify({"error": "Invalid credentials"}), 401

access_token = create_access_token(identity=user.id)
refresh_token = create_refresh_token(identity=user.id)

return jsonify({
    "access_token": access_token,
    "refresh_token": refresh_token,
    "user": user.to_dict()
})
@auth_bp.route('/refresh', methods=['POST']) @jwt_required(refresh=True) def refresh(): """Refresh access token.""" current_user_id = get_jwt_identity() access_token = create_access_token(identity=current_user_id)
return jsonify({"access_token": access_token})
@auth_bp.route('/me', methods=['GET']) @jwt_required() def get_current_user(): """Get current user from JWT.""" current_user_id = get_jwt_identity() user = User.query.get_or_404(current_user_id)
return jsonify(user.to_dict())
from flask_jwt_extended import ( create_access_token, create_refresh_token, jwt_required, get_jwt_identity )
@auth_bp.route('/login', methods=['POST']) def login(): """登录并返回JWT令牌。""" data = request.get_json()
user = User.query.filter_by(email=data['email']).first()
if not user or not user.check_password(data['password']):
    return jsonify({"error": "无效的凭据"}), 401

access_token = create_access_token(identity=user.id)
refresh_token = create_refresh_token(identity=user.id)

return jsonify({
    "access_token": access_token,
    "refresh_token": refresh_token,
    "user": user.to_dict()
})
@auth_bp.route('/refresh', methods=['POST']) @jwt_required(refresh=True) def refresh(): """刷新访问令牌。""" current_user_id = get_jwt_identity() access_token = create_access_token(identity=current_user_id)
return jsonify({"access_token": access_token})
@auth_bp.route('/me', methods=['GET']) @jwt_required() def get_current_user(): """从JWT获取当前用户。""" current_user_id = get_jwt_identity() user = User.query.get_or_404(current_user_id)
return jsonify(user.to_dict())

Protected route example

受保护路由示例

@users_bp.route('/int:user_id', methods=['DELETE']) @jwt_required() def delete_user(user_id): """Delete user (authenticated).""" current_user_id = get_jwt_identity()
# Authorization check
if current_user_id != user_id:
    return jsonify({"error": "Unauthorized"}), 403

user = User.query.get_or_404(user_id)
db.session.delete(user)
db.session.commit()

return '', 204
undefined
@users_bp.route('/int:user_id', methods=['DELETE']) @jwt_required() def delete_user(user_id): """删除用户（需认证）。""" current_user_id = get_jwt_identity()
# 权限检查
if current_user_id != user_id:
    return jsonify({"error": "无权限操作"}), 403

user = User.query.get_or_404(user_id)
db.session.delete(user)
db.session.commit()

return '', 204
undefined

Configuration Management

配置管理

Environment-Based Configuration

基于环境的配置

python
undefined
python
undefined

app/config.py

app/config.py

import os from datetime import timedelta
class Config: """Base configuration."""
# Flask
SECRET_KEY = os.getenv('SECRET_KEY', 'dev-secret-key-change-in-production')

# SQLAlchemy
SQLALCHEMY_DATABASE_URI = os.getenv('DATABASE_URL', 'sqlite:///app.db')
SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLALCHEMY_ECHO = False

# JWT
JWT_SECRET_KEY = os.getenv('JWT_SECRET_KEY', 'jwt-secret-change-in-production')
JWT_ACCESS_TOKEN_EXPIRES = timedelta(hours=1)
JWT_REFRESH_TOKEN_EXPIRES = timedelta(days=30)

# CORS
CORS_ORIGINS = os.getenv('CORS_ORIGINS', '*').split(',')

# Pagination
ITEMS_PER_PAGE = 20
MAX_ITEMS_PER_PAGE = 100

# Logging
LOG_LEVEL = os.getenv('LOG_LEVEL', 'INFO')
LOG_FILE = os.getenv('LOG_FILE', 'app.log')
class DevelopmentConfig(Config): """Development configuration.""" DEBUG = True TESTING = False SQLALCHEMY_ECHO = True
class TestingConfig(Config): """Testing configuration.""" TESTING = True SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:' WTF_CSRF_ENABLED = False
class ProductionConfig(Config): """Production configuration.""" DEBUG = False TESTING = False
# Strict security requirements
SECRET_KEY = os.getenv('SECRET_KEY')
JWT_SECRET_KEY = os.getenv('JWT_SECRET_KEY')

# Production database (PostgreSQL recommended)
SQLALCHEMY_DATABASE_URI = os.getenv('DATABASE_URL')

@classmethod
def init_app(cls, app):
    """Production-specific initialization."""
    # Log to syslog or external service
    import logging
    from logging.handlers import SysLogHandler

    syslog_handler = SysLogHandler()
    syslog_handler.setLevel(logging.WARNING)
    app.logger.addHandler(syslog_handler)
import os from datetime import timedelta
class Config: """基础配置。"""
# Flask
SECRET_KEY = os.getenv('SECRET_KEY', 'dev-secret-key-change-in-production')

# SQLAlchemy
SQLALCHEMY_DATABASE_URI = os.getenv('DATABASE_URL', 'sqlite:///app.db')
SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLALCHEMY_ECHO = False

# JWT
JWT_SECRET_KEY = os.getenv('JWT_SECRET_KEY', 'jwt-secret-change-in-production')
JWT_ACCESS_TOKEN_EXPIRES = timedelta(hours=1)
JWT_REFRESH_TOKEN_EXPIRES = timedelta(days=30)

# CORS
CORS_ORIGINS = os.getenv('CORS_ORIGINS', '*').split(',')

# 分页
ITEMS_PER_PAGE = 20
MAX_ITEMS_PER_PAGE = 100

# 日志
LOG_LEVEL = os.getenv('LOG_LEVEL', 'INFO')
LOG_FILE = os.getenv('LOG_FILE', 'app.log')
class DevelopmentConfig(Config): """开发环境配置。""" DEBUG = True TESTING = False SQLALCHEMY_ECHO = True
class TestingConfig(Config): """测试环境配置。""" TESTING = True SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:' WTF_CSRF_ENABLED = False
class ProductionConfig(Config): """生产环境配置。""" DEBUG = False TESTING = False
# 严格的安全要求
SECRET_KEY = os.getenv('SECRET_KEY')
JWT_SECRET_KEY = os.getenv('JWT_SECRET_KEY')

# 生产数据库（推荐PostgreSQL）
SQLALCHEMY_DATABASE_URI = os.getenv('DATABASE_URL')

@classmethod
def init_app(cls, app):
    """生产环境专属初始化。"""
    # 日志输出到syslog或外部服务
    import logging
    from logging.handlers import SysLogHandler

    syslog_handler = SysLogHandler()
    syslog_handler.setLevel(logging.WARNING)
    app.logger.addHandler(syslog_handler)

Configuration factory

配置工厂

config_by_name = { 'development': DevelopmentConfig, 'testing': TestingConfig, 'production': ProductionConfig, 'default': DevelopmentConfig }
def get_config(env_name='default'): """Get configuration by environment name.""" return config_by_name.get(env_name, DevelopmentConfig)
config_by_name = { 'development': DevelopmentConfig, 'testing': TestingConfig, 'production': ProductionConfig, 'default': DevelopmentConfig }
def get_config(env_name='default'): """根据环境名称获取配置。""" return config_by_name.get(env_name, DevelopmentConfig)

app/init.py

app/init.py

def create_app(config_name='default'): app = Flask(name) app.config.from_object(get_config(config_name))
return app
undefined
def create_app(config_name='default'): app = Flask(name) app.config.from_object(get_config(config_name))
return app
undefined

.env File Support

.env 文件支持

bash
undefined
bash
undefined

.env

.env

FLASK_APP=run.py FLASK_ENV=development SECRET_KEY=your-secret-key-here JWT_SECRET_KEY=your-jwt-secret-here DATABASE_URL=postgresql://user:password@localhost/dbname CORS_ORIGINS=http://localhost:3000,http://localhost:8000

```python
FLASK_APP=run.py FLASK_ENV=development SECRET_KEY=your-secret-key-here JWT_SECRET_KEY=your-jwt-secret-here DATABASE_URL=postgresql://user:password@localhost/dbname CORS_ORIGINS=http://localhost:3000,http://localhost:8000

```python

Load environment variables

加载环境变量

from dotenv import load_dotenv load_dotenv()
from dotenv import load_dotenv load_dotenv()

app/init.py

app/init.py

import os from dotenv import load_dotenv
def create_app(): load_dotenv()
config_name = os.getenv('FLASK_ENV', 'development')
app = Flask(__name__)
app.config.from_object(get_config(config_name))

return app
undefined
import os from dotenv import load_dotenv
def create_app(): load_dotenv()
config_name = os.getenv('FLASK_ENV', 'development')
app = Flask(__name__)
app.config.from_object(get_config(config_name))

return app
undefined

Error Handling and Logging

错误处理与日志

Global Error Handlers

全局错误处理器

python
undefined
python
undefined

app/errors/handlers.py

app/errors/handlers.py

from flask import jsonify from werkzeug.exceptions import HTTPException from sqlalchemy.exc import SQLAlchemyError
def register_error_handlers(app): """Register global error handlers."""
@app.errorhandler(404)
def not_found_error(error):
    return jsonify({
        "error": "Not Found",
        "message": "The requested resource was not found"
    }), 404

@app.errorhandler(400)
def bad_request_error(error):
    return jsonify({
        "error": "Bad Request",
        "message": str(error.description) if hasattr(error, 'description') else "Invalid request"
    }), 400

@app.errorhandler(401)
def unauthorized_error(error):
    return jsonify({
        "error": "Unauthorized",
        "message": "Authentication required"
    }), 401

@app.errorhandler(403)
def forbidden_error(error):
    return jsonify({
        "error": "Forbidden",
        "message": "You don't have permission to access this resource"
    }), 403

@app.errorhandler(500)
def internal_error(error):
    app.logger.error(f'Internal server error: {error}')
    db.session.rollback()
    return jsonify({
        "error": "Internal Server Error",
        "message": "An unexpected error occurred"
    }), 500

@app.errorhandler(HTTPException)
def handle_http_exception(error):
    """Handle all HTTP exceptions."""
    return jsonify({
        "error": error.name,
        "message": error.description
    }), error.code

@app.errorhandler(SQLAlchemyError)
def handle_db_error(error):
    """Handle database errors."""
    app.logger.error(f'Database error: {error}')
    db.session.rollback()
    return jsonify({
        "error": "Database Error",
        "message": "A database error occurred"
    }), 500
from flask import jsonify from werkzeug.exceptions import HTTPException from sqlalchemy.exc import SQLAlchemyError
def register_error_handlers(app): """注册全局错误处理器。"""
@app.errorhandler(404)
def not_found_error(error):
    return jsonify({
        "error": "未找到",
        "message": "请求的资源不存在"
    }), 404

@app.errorhandler(400)
def bad_request_error(error):
    return jsonify({
        "error": "错误请求",
        "message": str(error.description) if hasattr(error, 'description') else "无效请求"
    }), 400

@app.errorhandler(401)
def unauthorized_error(error):
    return jsonify({
        "error": "未授权",
        "message": "需要认证"
    }), 401

@app.errorhandler(403)
def forbidden_error(error):
    return jsonify({
        "error": "禁止访问",
        "message": "你没有权限访问该资源"
    }), 403

@app.errorhandler(500)
def internal_error(error):
    app.logger.error(f'内部服务器错误: {error}')
    db.session.rollback()
    return jsonify({
        "error": "内部服务器错误",
        "message": "发生了意外错误"
    }), 500

@app.errorhandler(HTTPException)
def handle_http_exception(error):
    """处理所有HTTP异常。"""
    return jsonify({
        "error": error.name,
        "message": error.description
    }), error.code

@app.errorhandler(SQLAlchemyError)
def handle_db_error(error):
    """处理数据库错误。"""
    app.logger.error(f'数据库错误: {error}')
    db.session.rollback()
    return jsonify({
        "error": "数据库错误",
        "message": "发生了数据库错误"
    }), 500

app/init.py

app/init.py

def create_app(): app = Flask(name)
from app.errors.handlers import register_error_handlers
register_error_handlers(app)

return app
undefined
def create_app(): app = Flask(name)
from app.errors.handlers import register_error_handlers
register_error_handlers(app)

return app
undefined

Logging Configuration

日志配置

python
undefined
python
undefined

app/logging_config.py

app/logging_config.py

import logging from logging.handlers import RotatingFileHandler import os
def configure_logging(app): """Configure application logging."""
if not app.debug and not app.testing:
    # Create logs directory if it doesn't exist
    if not os.path.exists('logs'):
        os.mkdir('logs')

    # File handler with rotation
    file_handler = RotatingFileHandler(
        'logs/flask_app.log',
        maxBytes=10240000,  # 10MB
        backupCount=10
    )
    file_handler.setFormatter(logging.Formatter(
        '%(asctime)s %(levelname)s: %(message)s '
        '[in %(pathname)s:%(lineno)d]'
    ))
    file_handler.setLevel(logging.INFO)
    app.logger.addHandler(file_handler)

    app.logger.setLevel(logging.INFO)
    app.logger.info('Flask application startup')
import logging from logging.handlers import RotatingFileHandler import os
def configure_logging(app): """配置应用日志。"""
if not app.debug and not app.testing:
    # 如果日志目录不存在则创建
    if not os.path.exists('logs'):
        os.mkdir('logs')

    # 带轮转的文件处理器
    file_handler = RotatingFileHandler(
        'logs/flask_app.log',
        maxBytes=10240000,  # 10MB
        backupCount=10
    )
    file_handler.setFormatter(logging.Formatter(
        '%(asctime)s %(levelname)s: %(message)s '
        '[in %(pathname)s:%(lineno)d]'
    ))
    file_handler.setLevel(logging.INFO)
    app.logger.addHandler(file_handler)

    app.logger.setLevel(logging.INFO)
    app.logger.info('Flask应用启动')

app/init.py

app/init.py

def create_app(): app = Flask(name)
from app.logging_config import configure_logging
configure_logging(app)

return app
def create_app(): app = Flask(name)
from app.logging_config import configure_logging
configure_logging(app)

return app

Usage in routes

在路由中使用

@users_bp.route('/', methods=['POST']) def create_user(): current_app.logger.info(f'Creating new user: {request.get_json()}') # ... create user ... current_app.logger.info(f'User created successfully: {user.id}')
undefined
@users_bp.route('/', methods=['POST']) def create_user(): current_app.logger.info(f'创建新用户: {request.get_json()}') # ... 创建用户 ... current_app.logger.info(f'用户创建成功: {user.id}')
undefined

Testing with pytest

使用pytest进行测试

Test Configuration

测试配置

python
undefined
python
undefined

conftest.py

conftest.py

import pytest from app import create_app from app.extensions import db from app.models.user import User
@pytest.fixture(scope='session') def app(): """Create application for testing.""" app = create_app('testing')
with app.app_context():
    db.create_all()
    yield app
    db.drop_all()
@pytest.fixture def client(app): """Flask test client.""" return app.test_client()
@pytest.fixture def runner(app): """Flask CLI test runner.""" return app.test_cli_runner()
@pytest.fixture def db_session(app): """Database session for testing.""" with app.app_context(): db.session.begin_nested() yield db.session db.session.rollback()
@pytest.fixture def sample_user(db_session): """Create sample user.""" user = User(email='test@example.com', name='Test User') user.set_password('password123') db_session.add(user) db_session.commit() return user
@pytest.fixture def auth_headers(client, sample_user): """Get JWT authentication headers.""" response = client.post('/auth/login', json={ 'email': 'test@example.com', 'password': 'password123' }) token = response.get_json()['access_token'] return {'Authorization': f'Bearer {token}'}
undefined
import pytest from app import create_app from app.extensions import db from app.models.user import User
@pytest.fixture(scope='session') def app(): """创建测试用应用。""" app = create_app('testing')
with app.app_context():
    db.create_all()
    yield app
    db.drop_all()
@pytest.fixture def client(app): """Flask测试客户端。""" return app.test_client()
@pytest.fixture def runner(app): """Flask CLI测试运行器。""" return app.test_cli_runner()
@pytest.fixture def db_session(app): """测试用数据库会话。""" with app.app_context(): db.session.begin_nested() yield db.session db.session.rollback()
@pytest.fixture def sample_user(db_session): """创建示例用户。""" user = User(email='test@example.com', name='Test User') user.set_password('password123') db_session.add(user) db_session.commit() return user
@pytest.fixture def auth_headers(client, sample_user): """获取JWT认证头。""" response = client.post('/auth/login', json={ 'email': 'test@example.com', 'password': 'password123' }) token = response.get_json()['access_token'] return {'Authorization': f'Bearer {token}'}
undefined

API Testing

API测试

python
undefined
python
undefined

tests/test_users.py

tests/test_users.py

import pytest from app.models.user import User
def test_create_user(client): """Test user creation endpoint.""" response = client.post('/api/users', json={ 'email': 'new@example.com', 'name': 'New User', 'password': 'password123' })
assert response.status_code == 201
data = response.get_json()
assert data['email'] == 'new@example.com'
assert data['name'] == 'New User'
assert 'id' in data
def test_get_user(client, sample_user): """Test get user endpoint.""" response = client.get(f'/api/users/{sample_user.id}')
assert response.status_code == 200
data = response.get_json()
assert data['id'] == sample_user.id
assert data['email'] == sample_user.email
def test_list_users(client, sample_user): """Test list users endpoint.""" response = client.get('/api/users')
assert response.status_code == 200
data = response.get_json()
assert 'users' in data
assert len(data['users']) > 0
def test_update_user(client, sample_user, auth_headers): """Test user update endpoint.""" response = client.put( f'/api/users/{sample_user.id}', json={'name': 'Updated Name'}, headers=auth_headers )
assert response.status_code == 200
data = response.get_json()
assert data['name'] == 'Updated Name'
def test_delete_user(client, sample_user, auth_headers): """Test user deletion endpoint.""" response = client.delete( f'/api/users/{sample_user.id}', headers=auth_headers )
assert response.status_code == 204
assert User.query.get(sample_user.id) is None
def test_authentication_required(client, sample_user): """Test that protected endpoints require authentication.""" response = client.delete(f'/api/users/{sample_user.id}') assert response.status_code == 401
undefined
import pytest from app.models.user import User
def test_create_user(client): """测试用户创建端点。""" response = client.post('/api/users', json={ 'email': 'new@example.com', 'name': 'New User', 'password': 'password123' })
assert response.status_code == 201
data = response.get_json()
assert data['email'] == 'new@example.com'
assert data['name'] == 'New User'
assert 'id' in data
def test_get_user(client, sample_user): """测试获取用户端点。""" response = client.get(f'/api/users/{sample_user.id}')
assert response.status_code == 200
data = response.get_json()
assert data['id'] == sample_user.id
assert data['email'] == sample_user.email
def test_list_users(client, sample_user): """测试列出用户端点。""" response = client.get('/api/users')
assert response.status_code == 200
data = response.get_json()
assert 'users' in data
assert len(data['users']) > 0
def test_update_user(client, sample_user, auth_headers): """测试用户更新端点。""" response = client.put( f'/api/users/{sample_user.id}', json={'name': 'Updated Name'}, headers=auth_headers )
assert response.status_code == 200
data = response.get_json()
assert data['name'] == 'Updated Name'
def test_delete_user(client, sample_user, auth_headers): """测试用户删除端点。""" response = client.delete( f'/api/users/{sample_user.id}', headers=auth_headers )
assert response.status_code == 204
assert User.query.get(sample_user.id) is None
def test_authentication_required(client, sample_user): """测试受保护端点需要认证。""" response = client.delete(f'/api/users/{sample_user.id}') assert response.status_code == 401
undefined

Deployment

部署

Production with Gunicorn

使用Gunicorn部署到生产环境

bash
undefined
bash
undefined

Install Gunicorn

安装Gunicorn

pip install gunicorn
pip install gunicorn

Run with Gunicorn

使用Gunicorn运行

gunicorn --workers 4 --bind 0.0.0.0:8000 "app:create_app()"
gunicorn --workers 4 --bind 0.0.0.0:8000 "app:create_app()"

With environment variable

带环境变量

gunicorn --workers 4 --bind 0.0.0.0:8000 --env FLASK_ENV=production "app:create_app()"

**gunicorn.conf.py:**
```python
import multiprocessing
gunicorn --workers 4 --bind 0.0.0.0:8000 --env FLASK_ENV=production "app:create_app()"

**gunicorn.conf.py:**
```python
import multiprocessing

Server socket

服务器套接字

bind = '0.0.0.0:8000' backlog = 2048
bind = '0.0.0.0:8000' backlog = 2048

Worker processes

工作进程

workers = multiprocessing.cpu_count() * 2 + 1 worker_class = 'sync' worker_connections = 1000 timeout = 30 keepalive = 2
workers = multiprocessing.cpu_count() * 2 + 1 worker_class = 'sync' worker_connections = 1000 timeout = 30 keepalive = 2

Logging

日志

accesslog = 'logs/access.log' errorlog = 'logs/error.log' loglevel = 'info'
accesslog = 'logs/access.log' errorlog = 'logs/error.log' loglevel = 'info'

Process naming

进程命名

proc_name = 'flask-app'
proc_name = 'flask-app'

Server mechanics

服务器机制

daemon = False pidfile = 'gunicorn.pid'
undefined
daemon = False pidfile = 'gunicorn.pid'
undefined

Docker Deployment

Docker部署

Dockerfile:
dockerfile
FROM python:3.11-slim

WORKDIR /app
Dockerfile:
dockerfile
FROM python:3.11-slim

WORKDIR /app

Install dependencies

安装依赖

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

Copy application

复制应用

COPY . .
COPY . .

Create non-root user

创建非root用户

RUN useradd -m -u 1000 flask && chown -R flask:flask /app USER flask
RUN useradd -m -u 1000 flask && chown -R flask:flask /app USER flask

Run with Gunicorn

使用Gunicorn运行

CMD ["gunicorn", "--config", "gunicorn.conf.py", "app:create_app()"]

**docker-compose.yml:**
```yaml
version: '3.8'

services:
  app:
    build: .
    ports:
      - "8000:8000"
    environment:
      - FLASK_ENV=production
      - DATABASE_URL=postgresql://user:password@db:5432/flask_db
      - SECRET_KEY=${SECRET_KEY}
      - JWT_SECRET_KEY=${JWT_SECRET_KEY}
    depends_on:
      - db
      - redis
    volumes:
      - ./logs:/app/logs

  db:
    image: postgres:15-alpine
    environment:
      - POSTGRES_USER=user
      - POSTGRES_PASSWORD=password
      - POSTGRES_DB=flask_db
    volumes:
      - postgres_data:/var/lib/postgresql/data

  redis:
    image: redis:7-alpine
    volumes:
      - redis_data:/data

  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf:ro
    depends_on:
      - app

volumes:
  postgres_data:
  redis_data:
CMD ["gunicorn", "--config", "gunicorn.conf.py", "app:create_app()"]

**docker-compose.yml:**
```yaml
version: '3.8'

services:
  app:
    build: .
    ports:
      - "8000:8000"
    environment:
      - FLASK_ENV=production
      - DATABASE_URL=postgresql://user:password@db:5432/flask_db
      - SECRET_KEY=${SECRET_KEY}
      - JWT_SECRET_KEY=${JWT_SECRET_KEY}
    depends_on:
      - db
      - redis
    volumes:
      - ./logs:/app/logs

  db:
    image: postgres:15-alpine
    environment:
      - POSTGRES_USER=user
      - POSTGRES_PASSWORD=password
      - POSTGRES_DB=flask_db
    volumes:
      - postgres_data:/var/lib/postgresql/data

  redis:
    image: redis:7-alpine
    volumes:
      - redis_data:/data

  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf:ro
    depends_on:
      - app

volumes:
  postgres_data:
  redis_data:

Microservices Patterns

微服务模式

Service Communication

服务通信

python
undefined
python
undefined

app/services/external_api.py

app/services/external_api.py

import requests from flask import current_app
class ExternalAPIService: """External API client service."""
def __init__(self, base_url, timeout=10):
    self.base_url = base_url
    self.timeout = timeout

def get(self, endpoint, **kwargs):
    """GET request to external API."""
    url = f"{self.base_url}/{endpoint}"
    try:
        response = requests.get(url, timeout=self.timeout, **kwargs)
        response.raise_for_status()
        return response.json()
    except requests.RequestException as e:
        current_app.logger.error(f'External API error: {e}')
        raise

def post(self, endpoint, data, **kwargs):
    """POST request to external API."""
    url = f"{self.base_url}/{endpoint}"
    try:
        response = requests.post(url, json=data, timeout=self.timeout, **kwargs)
        response.raise_for_status()
        return response.json()
    except requests.RequestException as e:
        current_app.logger.error(f'External API error: {e}')
        raise
import requests from flask import current_app
class ExternalAPIService: """外部API客户端服务。"""
def __init__(self, base_url, timeout=10):
    self.base_url = base_url
    self.timeout = timeout

def get(self, endpoint, **kwargs):
    """向外部API发送GET请求。"""
    url = f"{self.base_url}/{endpoint}"
    try:
        response = requests.get(url, timeout=self.timeout, **kwargs)
        response.raise_for_status()
        return response.json()
    except requests.RequestException as e:
        current_app.logger.error(f'外部API错误: {e}')
        raise

def post(self, endpoint, data, **kwargs):
    """向外部API发送POST请求。"""
    url = f"{self.base_url}/{endpoint}"
    try:
        response = requests.post(url, json=data, timeout=self.timeout, **kwargs)
        response.raise_for_status()
        return response.json()
    except requests.RequestException as e:
        current_app.logger.error(f'外部API错误: {e}')
        raise

Usage

使用示例

api_service = ExternalAPIService('https://api.example.com')
@app.route('/proxy/data') def proxy_data(): """Proxy request to external service.""" try: data = api_service.get('data') return jsonify(data) except Exception as e: return jsonify({"error": "External service unavailable"}), 503
undefined
api_service = ExternalAPIService('https://api.example.com')
@app.route('/proxy/data') def proxy_data(): """代理请求到外部服务。""" try: data = api_service.get('data') return jsonify(data) except Exception as e: return jsonify({"error": "外部服务不可用"}), 503
undefined

Health Checks

健康检查

python
undefined
python
undefined

app/routes/health.py

app/routes/health.py

from flask import Blueprint, jsonify from app.extensions import db from sqlalchemy import text
health_bp = Blueprint('health', name)
@health_bp.route('/health') def health_check(): """Basic health check.""" return jsonify({"status": "healthy"}), 200
@health_bp.route('/health/ready') def readiness_check(): """Readiness check with dependencies.""" checks = { "database": False, "status": "unhealthy" }
# Check database
try:
    db.session.execute(text('SELECT 1'))
    checks["database"] = True
except Exception as e:
    current_app.logger.error(f'Database health check failed: {e}')

# Overall status
checks["status"] = "healthy" if all([checks["database"]]) else "unhealthy"

status_code = 200 if checks["status"] == "healthy" else 503
return jsonify(checks), status_code
undefined
from flask import Blueprint, jsonify from app.extensions import db from sqlalchemy import text
health_bp = Blueprint('health', name)
@health_bp.route('/health') def health_check(): """基础健康检查。""" return jsonify({"status": "healthy"}), 200
@health_bp.route('/health/ready') def readiness_check(): """包含依赖检查的就绪性检查。""" checks = { "database": False, "status": "unhealthy" }
# 检查数据库
try:
    db.session.execute(text('SELECT 1'))
    checks["database"] = True
except Exception as e:
    current_app.logger.error(f'数据库健康检查失败: {e}')

# 整体状态
checks["status"] = "healthy" if all([checks["database"]]) else "unhealthy"

status_code = 200 if checks["status"] == "healthy" else 503
return jsonify(checks), status_code
undefined

Best Practices

最佳实践

1. Use Application Factory Pattern

1. 使用应用工厂模式

python
undefined
python
undefined

✅ GOOD: Application factory

✅ 推荐：应用工厂

def create_app(config_name='default'): app = Flask(name) app.config.from_object(get_config(config_name)) return app
def create_app(config_name='default'): app = Flask(name) app.config.from_object(get_config(config_name)) return app

❌ BAD: Global app instance

❌ 不推荐：全局应用实例

app = Flask(name)
undefined
app = Flask(name)
undefined

2. Use Blueprints for Modularity

2. 使用蓝图实现模块化

python
undefined
python
undefined

✅ GOOD: Organized blueprints

✅ 推荐：结构化蓝图

from app.routes.users import users_bp from app.routes.products import products_bp
app.register_blueprint(users_bp, url_prefix='/api/users') app.register_blueprint(products_bp, url_prefix='/api/products')
from app.routes.users import users_bp from app.routes.products import products_bp
app.register_blueprint(users_bp, url_prefix='/api/users') app.register_blueprint(products_bp, url_prefix='/api/products')

❌ BAD: All routes in single file

❌ 不推荐：所有路由放在单个文件中

undefined
undefined

3. Validate All Input

3. 验证所有输入

python
undefined
python
undefined

✅ GOOD: Validation with Marshmallow/Pydantic

✅ 推荐：使用Marshmallow/Pydantic验证

try: data = user_schema.load(request.get_json()) except ValidationError as e: return jsonify({"errors": e.messages}), 400
try: data = user_schema.load(request.get_json()) except ValidationError as e: return jsonify({"errors": e.messages}), 400

❌ BAD: No validation

❌ 不推荐：无验证

data = request.get_json() user = User(**data) # Unsafe!
undefined
data = request.get_json() user = User(**data) # 不安全!
undefined

4. Use Environment Variables for Secrets

4. 使用环境变量存储敏感信息

python
undefined
python
undefined

✅ GOOD: Environment variables

✅ 推荐：环境变量

SECRET_KEY = os.getenv('SECRET_KEY')
SECRET_KEY = os.getenv('SECRET_KEY')

❌ BAD: Hardcoded secrets

❌ 不推荐：硬编码敏感信息

SECRET_KEY = 'hardcoded-secret-key'
undefined
SECRET_KEY = 'hardcoded-secret-key'
undefined

5. Implement Proper Error Handling

5. 实现完善的错误处理

python
undefined
python
undefined

✅ GOOD: Global error handlers

✅ 推荐：全局错误处理器

@app.errorhandler(Exception) def handle_error(error): app.logger.error(f'Error: {error}') return jsonify({"error": "Internal server error"}), 500
@app.errorhandler(Exception) def handle_error(error): app.logger.error(f'错误: {error}') return jsonify({"error": "内部服务器错误"}), 500

❌ BAD: Unhandled exceptions

❌ 不推荐：未处理异常

undefined
undefined

Resources

资源

Related Skills

相关技能

When using Flask, consider these complementary skills:
  • pytest: Testing Flask applications with fixtures and test client
  • sqlalchemy: Database ORM patterns with Flask-SQLAlchemy integration
  • fastapi-local-dev: Modern async alternative for high-performance APIs
  • django: Batteries-included framework with built-in admin and ORM
使用Flask时，可考虑以下互补技能：
  • pytest: 使用fixture和测试客户端测试Flask应用
  • sqlalchemy: 结合Flask-SQLAlchemy的数据库ORM模式
  • fastapi-local-dev: 用于高性能API的现代异步替代方案
  • django: 自带管理后台和ORM的全功能框架

Quick Flask Testing Patterns (Inlined for Standalone Use)

快速Flask测试模式（独立使用）

python
undefined
python
undefined

Testing Flask with pytest

使用pytest测试Flask

import pytest from flask import Flask from app import create_app, db
@pytest.fixture def app(): """Create and configure test app""" app = create_app('testing') with app.app_context(): db.create_all() yield app db.drop_all()
@pytest.fixture def client(app): """Test client for making requests""" return app.test_client()
@pytest.fixture def runner(app): """CLI test runner""" return app.test_cli_runner()
import pytest from flask import Flask from app import create_app, db
@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 runner(app): """CLI测试运行器""" return app.test_cli_runner()

Test routes

测试路由

def test_home_page(client): response = client.get('/') assert response.status_code == 200 assert b'Welcome' in response.data
def test_api_endpoint(client): response = client.post('/api/users', json={ 'username': 'alice', 'email': 'alice@example.com' }) assert response.status_code == 201 data = response.get_json() assert data['username'] == 'alice'
def test_authentication(client): # Login response = client.post('/login', data={ 'username': 'alice', 'password': 'secret123' }) assert response.status_code == 302 # Redirect after login
# Access protected route
response = client.get('/dashboard')
assert response.status_code == 200
def test_home_page(client): response = client.get('/') assert response.status_code == 200 assert b'Welcome' in response.data
def test_api_endpoint(client): response = client.post('/api/users', json={ 'username': 'alice', 'email': 'alice@example.com' }) assert response.status_code == 201 data = response.get_json() assert data['username'] == 'alice'
def test_authentication(client): # 登录 response = client.post('/login', data={ 'username': 'alice', 'password': 'secret123' }) assert response.status_code == 302 # 登录后重定向
# 访问受保护路由
response = client.get('/dashboard')
assert response.status_code == 200

Test with database

数据库测试

def test_user_creation(app): with app.app_context(): user = User(username='bob', email='bob@example.com') db.session.add(user) db.session.commit()
    found = User.query.filter_by(username='bob').first()
    assert found is not None
    assert found.email == 'bob@example.com'
def test_user_creation(app): with app.app_context(): user = User(username='bob', email='bob@example.com') db.session.add(user) db.session.commit()
    found = User.query.filter_by(username='bob').first()
    assert found is not None
    assert found.email == 'bob@example.com'

Test error handling

错误处理测试

def test_404_error(client): response = client.get('/nonexistent') assert response.status_code == 404 assert b'Not Found' in response.data
undefined
def test_404_error(client): response = client.get('/nonexistent') assert response.status_code == 404 assert b'Not Found' in response.data
undefined

Quick SQLAlchemy Integration (Inlined for Standalone Use)

快速SQLAlchemy集成（独立使用）

python
undefined
python
undefined

Flask-SQLAlchemy setup

Flask-SQLAlchemy设置

from flask import Flask from flask_sqlalchemy import SQLAlchemy from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase): pass
db = SQLAlchemy(model_class=Base)
def create_app(): app = Flask(name) app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://user:pass@localhost/db' app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db.init_app(app)

with app.app_context():
    db.create_all()

return app
from flask import Flask from flask_sqlalchemy import SQLAlchemy from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase): pass
db = SQLAlchemy(model_class=Base)
def create_app(): app = Flask(name) app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://user:pass@localhost/db' app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db.init_app(app)

with app.app_context():
    db.create_all()

return app

Define models

定义模型

class User(db.Model): tablename = 'users'
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(50), unique=True, nullable=False)
email = db.Column(db.String(100), unique=True, nullable=False)
posts = db.relationship('Post', backref='author', lazy='dynamic')
class Post(db.Model): tablename = 'posts'
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(200), nullable=False)
content = db.Column(db.Text, nullable=False)
user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
class User(db.Model): tablename = 'users'
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(50), unique=True, nullable=False)
email = db.Column(db.String(100), unique=True, nullable=False)
posts = db.relationship('Post', backref='author', lazy='dynamic')
class Post(db.Model): tablename = 'posts'
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(200), nullable=False)
content = db.Column(db.Text, nullable=False)
user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)

Query patterns

查询模式

@app.route('/users/int:user_id') def get_user(user_id): user = db.session.get(User, user_id) if not user: abort(404) return jsonify({ 'id': user.id, 'username': user.username, 'email': user.email, 'posts': [{'title': p.title} for p in user.posts] })
@app.route('/users', methods=['POST']) def create_user(): data = request.get_json() user = User(username=data['username'], email=data['email']) db.session.add(user) db.session.commit() return jsonify({'id': user.id}), 201
@app.route('/users/int:user_id') def get_user(user_id): user = db.session.get(User, user_id) if not user: abort(404) return jsonify({ 'id': user.id, 'username': user.username, 'email': user.email, 'posts': [{'title': p.title} for p in user.posts] })
@app.route('/users', methods=['POST']) def create_user(): data = request.get_json() user = User(username=data['username'], email=data['email']) db.session.add(user) db.session.commit() return jsonify({'id': user.id}), 201

Transaction handling

事务处理

from sqlalchemy.exc import IntegrityError
@app.route('/transfer', methods=['POST']) def transfer_funds(): try: # All operations in single transaction sender = User.query.get_or_404(request.json['sender_id']) receiver = User.query.get_or_404(request.json['receiver_id']) amount = request.json['amount']
    sender.balance -= amount
    receiver.balance += amount

    db.session.commit()
    return jsonify({'status': 'success'})
except IntegrityError:
    db.session.rollback()
    return jsonify({'error': 'Transaction failed'}), 400
undefined
from sqlalchemy.exc import IntegrityError
@app.route('/transfer', methods=['POST']) def transfer_funds(): try: # 所有操作在单个事务中 sender = User.query.get_or_404(request.json['sender_id']) receiver = User.query.get_or_404(request.json['receiver_id']) amount = request.json['amount']
    sender.balance -= amount
    receiver.balance += amount

    db.session.commit()
    return jsonify({'status': 'success'})
except IntegrityError:
    db.session.rollback()
    return jsonify({'error': '交易失败'}), 400
undefined

Quick FastAPI Comparison (Inlined for Standalone Use)

快速FastAPI对比（独立使用）

When to Choose FastAPI over Flask:
  • Need async/await for high concurrency
  • Want automatic API documentation (OpenAPI/Swagger)
  • Require built-in data validation (Pydantic)
  • Building modern microservices or GraphQL APIs
  • Need WebSocket or Server-Sent Events support
When to Stick with Flask:
  • Building traditional server-rendered web apps
  • Existing large Flask codebase
  • Need mature ecosystem and extensive plugins
  • Team familiar with synchronous Python patterns
  • Simpler deployment (no async runtime complexity)
Migration Considerations:
python
undefined
何时选择FastAPI而非Flask：
  • 需要async/await实现高并发
  • 想要自动生成API文档（OpenAPI/Swagger）
  • 需要内置数据验证（Pydantic）
  • 构建现代微服务或GraphQL API
  • 需要WebSocket或Server-Sent Events支持
何时坚持使用Flask：
  • 构建传统服务器渲染的Web应用
  • 已有大型Flask代码库
  • 需要成熟的生态系统和丰富的插件
  • 团队熟悉同步Python模式
  • 部署更简单（无异步运行时复杂度）
迁移注意事项：
python
undefined

Flask pattern

Flask模式

@app.route('/users/int:user_id') def get_user(user_id): user = User.query.get_or_404(user_id) return jsonify({'id': user.id, 'name': user.name})
@app.route('/users/int:user_id') def get_user(user_id): user = User.query.get_or_404(user_id) return jsonify({'id': user.id, 'name': user.name})

FastAPI equivalent

FastAPI等效实现

@app.get('/users/{user_id}', response_model=UserResponse) async def get_user(user_id: int, db: AsyncSession = Depends(get_db)): user = await db.get(User, user_id) if not user: raise HTTPException(status_code=404, detail="User not found") return user

[Full pytest, SQLAlchemy, and FastAPI patterns available in respective skills if deployed together]
@app.get('/users/{user_id}', response_model=UserResponse) async def get_user(user_id: int, db: AsyncSession = Depends(get_db)): user = await db.get(User, user_id) if not user: raise HTTPException(status_code=404, detail="用户未找到") return user

[完整的pytest、SQLAlchemy和FastAPI模式可在相关技能中获取（需一起部署）]