flask

Flask - 轻量级 Python Web 框架

概述

Flask 是一个用于 Python Web 开发的微框架，专为构建微服务、REST API 和灵活的 Web 应用而设计。其精简的核心和丰富的扩展生态系统，使其非常适合需要轻量架构、快速开发并能完全控制组件的项目。

主要功能：

• 微框架理念（核心精简、可扩展）
• 使用 Flask-RESTful 开发 API
• 使用 Blueprint 构建模块化应用结构
• 通过 Flask-SQLAlchemy 集成 SQLAlchemy
• Jinja2 模板引擎
• 内置开发服务器，支持自动重新加载
• 基于 Werkzeug WSGI 工具包
• 庞大的扩展生态系统（Flask-Login、Flask-JWT、Flask-CORS）
• 使用 Gunicorn/uWSGI 部署到生产环境

安装：

# Basic Flask
pip install flask

# Flask with common extensions
pip install flask flask-restful flask-sqlalchemy flask-login flask-cors

# With database support
pip install flask flask-sqlalchemy psycopg2-binary  # PostgreSQL

# Full microservices stack
pip install flask flask-restful marshmallow flask-jwt-extended redis

基本应用模式

1. 最简 Flask 应用

# 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)

运行：

# Development server
python app.py

# Or using flask CLI
export FLASK_APP=app.py
export FLASK_ENV=development
flask run

# Custom port
flask run --port 8000 --host 0.0.0.0

2. 应用工厂模式

建议用于生产环境和测试：

# 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

# 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()

# 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

# run.py
from app import create_app

app = create_app()

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

运行：

export FLASK_APP=run.py
flask run

3. 请求和响应处理

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

# 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

Blueprint - 模块化应用

Blueprint 结构

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

Blueprint 实现

# 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

# 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

使用 Flask-RESTful 构建 API

基本 REST API

# 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

# 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

# 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

请求验证

Marshmallow Schema

# 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))

# 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))

Pydantic 验证

# 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

# 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

SQLAlchemy 集成

使用 Flask-SQLAlchemy 定义模型

# 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}>'

# 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()
        }

# 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
    })

数据库迁移

# Initialize migrations
flask db init

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

# Apply migration
flask db upgrade

# Rollback
flask db downgrade

# migrations/versions/xxx_create_users.py (auto-generated)
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')

身份验证

基于 Flask-Login 会话的身份验证

# app/extensions.py
from flask_login import LoginManager

login_manager = LoginManager()

# 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

# app/models/user.py
from flask_login import UserMixin

class User(UserMixin, db.Model):
    # ... existing fields ...

    def get_id(self):
        return str(self.id)

# 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())

JWT 身份验证

# app/extensions.py
from flask_jwt_extended import JWTManager

jwt = JWTManager()

# 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

# 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())

# 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

配置管理

基于环境的配置

# 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)

# 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)

# app/__init__.py
def create_app(config_name='default'):
    app = Flask(__name__)
    app.config.from_object(get_config(config_name))

    return app

.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

# Load environment variables
from dotenv import load_dotenv
load_dotenv()

# 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

错误处理和日志记录

全局错误处理器

# 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

# app/__init__.py
def create_app():
    app = Flask(__name__)

    from app.errors.handlers import register_error_handlers
    register_error_handlers(app)

    return app

日志配置

# 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')

# app/__init__.py
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}')

使用 pytest 进行测试

测试配置

# 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='[email protected]', 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': '[email protected]',
        'password': 'password123'
    })
    token = response.get_json()['access_token']
    return {'Authorization': f'Bearer {token}'}

API 测试

# 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': '[email protected]',
        'name': 'New User',
        'password': 'password123'
    })

    assert response.status_code == 201
    data = response.get_json()
    assert data['email'] == '[email protected]'
    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

部署

使用 Gunicorn 部署到生产环境

# Install Gunicorn
pip install gunicorn

# Run with Gunicorn
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：

import multiprocessing

# Server socket
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

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

# Process naming
proc_name = 'flask-app'

# Server mechanics
daemon = False
pidfile = 'gunicorn.pid'

Docker 部署

Dockerfile：

FROM python:3.11-slim

WORKDIR /app

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

# Copy application
COPY . .

# Create non-root user
RUN useradd -m -u 1000 flask && chown -R flask:flask /app
USER flask

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

docker-compose.yml：

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:

微服务模式

服务间通信

# 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

# 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

健康检查

# 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

最佳实践

1. 使用应用工厂模式

# ✅ GOOD: Application factory
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__)

2. 使用 Blueprint 实现模块化

# ✅ 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')

# ❌ BAD: All routes in single file

3. 验证所有输入

# ✅ GOOD: Validation with Marshmallow/Pydantic
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!

4. 使用环境变量保存机密信息

# ✅ GOOD: Environment variables
SECRET_KEY = os.getenv('SECRET_KEY')

# ❌ BAD: Hardcoded secrets
SECRET_KEY = 'hardcoded-secret-key'

5. 实现适当的错误处理

# ✅ GOOD: Global error handlers
@app.errorhandler(Exception)
def handle_error(error):
    app.logger.error(f'Error: {error}')
    return jsonify({"error": "Internal server error"}), 500

# ❌ BAD: Unhandled exceptions

资源

• 官方文档：https://flask.palletsprojects.com/
• Flask-RESTful：https://flask-restful.readthedocs.io/
• Flask-SQLAlchemy：https://flask-sqlalchemy.palletsprojects.com/
• Flask-JWT-Extended：https://flask-jwt-extended.readthedocs.io/
• Marshmallow：https://marshmallow.readthedocs.io/
• Pydantic：https://pydantic-docs.helpmanual.io/

相关技能

使用 Flask 时，可考虑搭配以下技能：

• pytest：使用 fixture 和测试客户端测试 Flask 应用
• sqlalchemy：通过 Flask-SQLAlchemy 集成实现数据库 ORM 模式
• fastapi-local-dev：适用于高性能 API 的现代异步替代方案
• django：内置管理后台和 ORM 的全功能框架

Flask 快速测试模式（内联，可独立使用）

# Testing Flask with pytest
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()

# 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': '[email protected]'
    })
    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

# Test with database
def test_user_creation(app):
    with app.app_context():
        user = User(username='bob', email='[email protected]')
        db.session.add(user)
        db.session.commit()

        found = User.query.filter_by(username='bob').first()
        assert found is not None
        assert found.email == '[email protected]'

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

SQLAlchemy 快速集成（内联，可独立使用）

# Flask-SQLAlchemy setup
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)

# 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

# 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

FastAPI 与 Flask 的快速对比（内嵌版本，可独立使用）

适合选择 FastAPI 而非 Flask 的情况：

• 需要使用 async/await 实现高并发
• 希望自动生成 API 文档（OpenAPI/Swagger）
• 需要内置数据验证（Pydantic）
• 正在构建现代微服务或 GraphQL API
• 需要支持 WebSocket 或 Server-Sent Events

适合继续使用 Flask 的情况：

• 正在构建传统的服务端渲染 Web 应用
• 已有大型 Flask 代码库
• 需要成熟的生态系统和丰富的插件
• 团队熟悉同步 Python 模式
• 希望简化部署（无需处理异步运行时的复杂性）

迁移注意事项：

# Flask pattern
@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
@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

[如果同时部署，相关技能中提供了完整的 pytest、SQLAlchemy 和 FastAPI 模式]