Back to Details

flask-api-development

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

Flask API Development

Flask API 开发

Overview

概述

Create efficient Flask APIs with blueprints for modular organization, SQLAlchemy for ORM, JWT authentication, comprehensive error handling, and proper request validation following REST principles.
创建高效的Flask API，使用Blueprint实现模块化组织，SQLAlchemy作为ORM框架，JWT认证，全面的错误处理，并遵循REST原则进行规范的请求验证。

When to Use

适用场景

  • Building RESTful APIs with Flask
  • Creating microservices with minimal overhead
  • Implementing lightweight authentication systems
  • Designing API endpoints with proper validation
  • Integrating with relational databases
  • Building request/response handling systems
  • 使用Flask构建RESTful API
  • 创建低开销的微服务
  • 实现轻量级认证系统
  • 设计带规范验证的API端点
  • 集成关系型数据库
  • 构建请求/响应处理系统

Instructions

操作步骤

1. Flask Application Setup

1. Flask 应用设置

python
undefined
python
undefined

app.py

app.py

from flask import Flask, request, jsonify from flask_cors import CORS from flask_sqlalchemy import SQLAlchemy from flask_jwt_extended import JWTManager import os
app = Flask(name) app.config['SQLALCHEMY_DATABASE_URI'] = os.getenv('DATABASE_URL', 'sqlite:///app.db') app.config['JWT_SECRET_KEY'] = os.getenv('JWT_SECRET_KEY', 'dev-secret') app.config['JSON_SORT_KEYS'] = False
db = SQLAlchemy(app) jwt = JWTManager(app) CORS(app)
from flask import Flask, request, jsonify from flask_cors import CORS from flask_sqlalchemy import SQLAlchemy from flask_jwt_extended import JWTManager import os
app = Flask(name) app.config['SQLALCHEMY_DATABASE_URI'] = os.getenv('DATABASE_URL', 'sqlite:///app.db') app.config['JWT_SECRET_KEY'] = os.getenv('JWT_SECRET_KEY', 'dev-secret') app.config['JSON_SORT_KEYS'] = False
db = SQLAlchemy(app) jwt = JWTManager(app) CORS(app)

Request ID middleware

Request ID 中间件

@app.before_request def assign_request_id(): import uuid request.request_id = str(uuid.uuid4())
@app.before_request def assign_request_id(): import uuid request.request_id = str(uuid.uuid4())

Error handlers

错误处理器

@app.errorhandler(400) def bad_request(error): return jsonify({ 'error': 'Bad Request', 'message': str(error), 'request_id': request.request_id }), 400
@app.errorhandler(404) def not_found(error): return jsonify({ 'error': 'Not Found', 'message': 'Resource does not exist', 'request_id': request.request_id }), 404
@app.errorhandler(500) def internal_error(error): db.session.rollback() return jsonify({ 'error': 'Internal Server Error', 'request_id': request.request_id }), 500
if name == 'main': app.run(debug=os.getenv('ENV') != 'production')
undefined
@app.errorhandler(400) def bad_request(error): return jsonify({ 'error': 'Bad Request', 'message': str(error), 'request_id': request.request_id }), 400
@app.errorhandler(404) def not_found(error): return jsonify({ 'error': 'Not Found', 'message': 'Resource does not exist', 'request_id': request.request_id }), 404
@app.errorhandler(500) def internal_error(error): db.session.rollback() return jsonify({ 'error': 'Internal Server Error', 'request_id': request.request_id }), 500
if name == 'main': app.run(debug=os.getenv('ENV') != 'production')
undefined

2. Database Models with SQLAlchemy

2. 使用SQLAlchemy定义数据库模型

python
undefined
python
undefined

models.py

models.py

from datetime import datetime from flask_sqlalchemy import SQLAlchemy from sqlalchemy.dialects.postgresql import UUID import uuid
db = SQLAlchemy()
class User(db.Model): tablename = 'users'
id = db.Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
email = db.Column(db.String(255), unique=True, nullable=False, index=True)
password_hash = db.Column(db.String(255), nullable=False)
first_name = db.Column(db.String(100))
last_name = db.Column(db.String(100))
role = db.Column(db.String(20), default='user', index=True)
is_active = db.Column(db.Boolean, default=True)
created_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)
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 __repr__(self):
    return f'<User {self.email}>'

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

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

def to_dict(self):
    return {
        'id': str(self.id),
        'email': self.email,
        'first_name': self.first_name,
        'last_name': self.last_name,
        'role': self.role,
        'created_at': self.created_at.isoformat()
    }
class Post(db.Model): tablename = 'posts'
id = db.Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
title = db.Column(db.String(255), nullable=False, index=True)
content = db.Column(db.Text, nullable=False)
published = db.Column(db.Boolean, default=False)
user_id = db.Column(UUID(as_uuid=True), db.ForeignKey('users.id'), nullable=False)
created_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

def to_dict(self):
    return {
        'id': str(self.id),
        'title': self.title,
        'content': self.content,
        'published': self.published,
        'author_id': str(self.user_id),
        'created_at': self.created_at.isoformat()
    }
undefined
from datetime import datetime from flask_sqlalchemy import SQLAlchemy from sqlalchemy.dialects.postgresql import UUID import uuid
db = SQLAlchemy()
class User(db.Model): tablename = 'users'
id = db.Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
email = db.Column(db.String(255), unique=True, nullable=False, index=True)
password_hash = db.Column(db.String(255), nullable=False)
first_name = db.Column(db.String(100))
last_name = db.Column(db.String(100))
role = db.Column(db.String(20), default='user', index=True)
is_active = db.Column(db.Boolean, default=True)
created_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)
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 __repr__(self):
    return f'<User {self.email}>'

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

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

def to_dict(self):
    return {
        'id': str(self.id),
        'email': self.email,
        'first_name': self.first_name,
        'last_name': self.last_name,
        'role': self.role,
        'created_at': self.created_at.isoformat()
    }
class Post(db.Model): tablename = 'posts'
id = db.Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
title = db.Column(db.String(255), nullable=False, index=True)
content = db.Column(db.Text, nullable=False)
published = db.Column(db.Boolean, default=False)
user_id = db.Column(UUID(as_uuid=True), db.ForeignKey('users.id'), nullable=False)
created_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

def to_dict(self):
    return {
        'id': str(self.id),
        'title': self.title,
        'content': self.content,
        'published': self.published,
        'author_id': str(self.user_id),
        'created_at': self.created_at.isoformat()
    }
undefined

3. Authentication and JWT

3. 认证与JWT

python
undefined
python
undefined

auth.py

auth.py

from flask import request, jsonify from flask_jwt_extended import create_access_token, jwt_required, get_jwt_identity from functools import wraps from models import User, db
def authenticate_user(email, password): user = User.query.filter_by(email=email).first() if user and user.verify_password(password): return user return None
def login_required(f): @wraps(f) @jwt_required() def decorated_function(*args, **kwargs): identity = get_jwt_identity() user = User.query.get(identity) if not user or not user.is_active: return jsonify({'error': 'User not found or inactive'}), 401 request.current_user = user return f(*args, **kwargs) return decorated_function
def admin_required(f): @wraps(f) @login_required def decorated_function(*args, **kwargs): if request.current_user.role != 'admin': return jsonify({'error': 'Admin access required'}), 403 return f(*args, **kwargs) return decorated_function
from flask import request, jsonify from flask_jwt_extended import create_access_token, jwt_required, get_jwt_identity from functools import wraps from models import User, db
def authenticate_user(email, password): user = User.query.filter_by(email=email).first() if user and user.verify_password(password): return user return None
def login_required(f): @wraps(f) @jwt_required() def decorated_function(*args, **kwargs): identity = get_jwt_identity() user = User.query.get(identity) if not user or not user.is_active: return jsonify({'error': 'User not found or inactive'}), 401 request.current_user = user return f(*args, **kwargs) return decorated_function
def admin_required(f): @wraps(f) @login_required def decorated_function(*args, **kwargs): if request.current_user.role != 'admin': return jsonify({'error': 'Admin access required'}), 403 return f(*args, **kwargs) return decorated_function

routes/auth.py

routes/auth.py

from flask import Blueprint, request, jsonify from auth import authenticate_user, login_required from models import User, db from flask_jwt_extended import create_access_token
auth_bp = Blueprint('auth', name, url_prefix='/api/auth')
@auth_bp.route('/login', methods=['POST']) def login(): data = request.get_json() if not data or not data.get('email') or not data.get('password'): return jsonify({'error': 'Missing credentials'}), 400
user = authenticate_user(data['email'], data['password'])
if not user:
    return jsonify({'error': 'Invalid credentials'}), 401

access_token = create_access_token(identity=str(user.id))
return jsonify({
    'access_token': access_token,
    'user': user.to_dict()
}), 200
@auth_bp.route('/register', methods=['POST']) def register(): data = request.get_json() if User.query.filter_by(email=data['email']).first(): return jsonify({'error': 'Email already exists'}), 409
user = User(email=data['email'], first_name=data.get('first_name'))
user.set_password(data['password'])
db.session.add(user)
db.session.commit()

return jsonify({'user': user.to_dict()}), 201
@auth_bp.route('/profile', methods=['GET']) @login_required def get_profile(): return jsonify({'user': request.current_user.to_dict()}), 200
undefined
from flask import Blueprint, request, jsonify from auth import authenticate_user, login_required from models import User, db from flask_jwt_extended import create_access_token
auth_bp = Blueprint('auth', name, url_prefix='/api/auth')
@auth_bp.route('/login', methods=['POST']) def login(): data = request.get_json() if not data or not data.get('email') or not data.get('password'): return jsonify({'error': 'Missing credentials'}), 400
user = authenticate_user(data['email'], data['password'])
if not user:
    return jsonify({'error': 'Invalid credentials'}), 401

access_token = create_access_token(identity=str(user.id))
return jsonify({
    'access_token': access_token,
    'user': user.to_dict()
}), 200
@auth_bp.route('/register', methods=['POST']) def register(): data = request.get_json() if User.query.filter_by(email=data['email']).first(): return jsonify({'error': 'Email already exists'}), 409
user = User(email=data['email'], first_name=data.get('first_name'))
user.set_password(data['password'])
db.session.add(user)
db.session.commit()

return jsonify({'user': user.to_dict()}), 201
@auth_bp.route('/profile', methods=['GET']) @login_required def get_profile(): return jsonify({'user': request.current_user.to_dict()}), 200
undefined

4. Blueprints for Modular API Design

4. 使用Blueprint实现模块化API设计

python
undefined
python
undefined

routes/users.py

routes/users.py

from flask import Blueprint, request, jsonify from auth import login_required, admin_required from models import User, db from sqlalchemy import or_
users_bp = Blueprint('users', name, url_prefix='/api/users')
@users_bp.route('', methods=['GET']) @login_required def list_users(): page = request.args.get('page', 1, type=int) limit = request.args.get('limit', 20, type=int) search = request.args.get('q', '', type=str)
query = User.query
if search:
    query = query.filter(or_(
        User.email.ilike(f'%{search}%'),
        User.first_name.ilike(f'%{search}%')
    ))

paginated = query.paginate(page=page, per_page=limit)
return jsonify({
    'data': [user.to_dict() for user in paginated.items],
    'pagination': {
        'page': page,
        'limit': limit,
        'total': paginated.total,
        'pages': paginated.pages
    }
}), 200
@users_bp.route('/<user_id>', methods=['GET']) @login_required def get_user(user_id): user = User.query.get(user_id) if not user: return jsonify({'error': 'User not found'}), 404 return jsonify({'user': user.to_dict()}), 200
@users_bp.route('/<user_id>', methods=['PATCH']) @login_required def update_user(user_id): if str(request.current_user.id) != user_id: return jsonify({'error': 'Unauthorized'}), 403
user = User.query.get(user_id)
if not user:
    return jsonify({'error': 'User not found'}), 404

data = request.get_json()
if 'first_name' in data:
    user.first_name = data['first_name']
if 'last_name' in data:
    user.last_name = data['last_name']

db.session.commit()
return jsonify({'user': user.to_dict()}), 200
@users_bp.route('/<user_id>', methods=['DELETE']) @admin_required def delete_user(user_id): user = User.query.get(user_id) if not user: return jsonify({'error': 'User not found'}), 404
db.session.delete(user)
db.session.commit()
return '', 204
undefined
from flask import Blueprint, request, jsonify from auth import login_required, admin_required from models import User, db from sqlalchemy import or_
users_bp = Blueprint('users', name, url_prefix='/api/users')
@users_bp.route('', methods=['GET']) @login_required def list_users(): page = request.args.get('page', 1, type=int) limit = request.args.get('limit', 20, type=int) search = request.args.get('q', '', type=str)
query = User.query
if search:
    query = query.filter(or_(
        User.email.ilike(f'%{search}%'),
        User.first_name.ilike(f'%{search}%')
    ))

paginated = query.paginate(page=page, per_page=limit)
return jsonify({
    'data': [user.to_dict() for user in paginated.items],
    'pagination': {
        'page': page,
        'limit': limit,
        'total': paginated.total,
        'pages': paginated.pages
    }
}), 200
@users_bp.route('/<user_id>', methods=['GET']) @login_required def get_user(user_id): user = User.query.get(user_id) if not user: return jsonify({'error': 'User not found'}), 404 return jsonify({'user': user.to_dict()}), 200
@users_bp.route('/<user_id>', methods=['PATCH']) @login_required def update_user(user_id): if str(request.current_user.id) != user_id: return jsonify({'error': 'Unauthorized'}), 403
user = User.query.get(user_id)
if not user:
    return jsonify({'error': 'User not found'}), 404

data = request.get_json()
if 'first_name' in data:
    user.first_name = data['first_name']
if 'last_name' in data:
    user.last_name = data['last_name']

db.session.commit()
return jsonify({'user': user.to_dict()}), 200
@users_bp.route('/<user_id>', methods=['DELETE']) @admin_required def delete_user(user_id): user = User.query.get(user_id) if not user: return jsonify({'error': 'User not found'}), 404
db.session.delete(user)
db.session.commit()
return '', 204
undefined

5. Request Validation

5. 请求验证

python
undefined
python
undefined

validators.py

validators.py

from flask import request, jsonify from functools import wraps
def validate_json(*required_fields): def decorator(f): @wraps(f) def decorated_function(*args, **kwargs): if not request.is_json: return jsonify({'error': 'Request body must be JSON'}), 400
        data = request.get_json()
        missing = [field for field in required_fields if field not in data]

        if missing:
            return jsonify({
                'error': 'Missing required fields',
                'missing_fields': missing
            }), 400

        return f(*args, **kwargs)
    return decorated_function
return decorator
def validate_email(email): import re pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+.[a-zA-Z]{2,}$' return re.match(pattern, email) is not None
from flask import request, jsonify from functools import wraps
def validate_json(*required_fields): def decorator(f): @wraps(f) def decorated_function(*args, **kwargs): if not request.is_json: return jsonify({'error': 'Request body must be JSON'}), 400
        data = request.get_json()
        missing = [field for field in required_fields if field not in data]

        if missing:
            return jsonify({
                'error': 'Missing required fields',
                'missing_fields': missing
            }), 400

        return f(*args, **kwargs)
    return decorated_function
return decorator
def validate_email(email): import re pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+.[a-zA-Z]{2,}$' return re.match(pattern, email) is not None

Usage

使用示例

@users_bp.route('', methods=['POST']) @validate_json('email', 'password', 'first_name') def create_user(): data = request.get_json() if not validate_email(data['email']): return jsonify({'error': 'Invalid email format'}), 400 # ... rest of logic
undefined
@users_bp.route('', methods=['POST']) @validate_json('email', 'password', 'first_name') def create_user(): data = request.get_json() if not validate_email(data['email']): return jsonify({'error': 'Invalid email format'}), 400 # ... 剩余逻辑
undefined

6. Application Factory and Configuration

6. 应用工厂与配置

python
undefined
python
undefined

config.py

config.py

import os
class Config: SQLALCHEMY_TRACK_MODIFICATIONS = False JSON_SORT_KEYS = False
class DevelopmentConfig(Config): DEBUG = True TESTING = False SQLALCHEMY_DATABASE_URI = 'sqlite:///app.db'
class ProductionConfig(Config): DEBUG = False TESTING = False SQLALCHEMY_DATABASE_URI = os.getenv('DATABASE_URL') JWT_SECRET_KEY = os.getenv('JWT_SECRET_KEY')
class TestingConfig(Config): TESTING = True SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:'
import os
class Config: SQLALCHEMY_TRACK_MODIFICATIONS = False JSON_SORT_KEYS = False
class DevelopmentConfig(Config): DEBUG = True TESTING = False SQLALCHEMY_DATABASE_URI = 'sqlite:///app.db'
class ProductionConfig(Config): DEBUG = False TESTING = False SQLALCHEMY_DATABASE_URI = os.getenv('DATABASE_URL') JWT_SECRET_KEY = os.getenv('JWT_SECRET_KEY')
class TestingConfig(Config): TESTING = True SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:'

factory.py

factory.py

from flask import Flask from flask_sqlalchemy import SQLAlchemy from flask_jwt_extended import JWTManager
def create_app(config_name='development'): app = Flask(name)
if config_name == 'production':
    from config import ProductionConfig
    app.config.from_object(ProductionConfig)
else:
    from config import DevelopmentConfig
    app.config.from_object(DevelopmentConfig)

db = SQLAlchemy(app)
jwt = JWTManager(app)

# Register blueprints
from routes.auth import auth_bp
from routes.users import users_bp
app.register_blueprint(auth_bp)
app.register_blueprint(users_bp)

return app
undefined
from flask import Flask from flask_sqlalchemy import SQLAlchemy from flask_jwt_extended import JWTManager
def create_app(config_name='development'): app = Flask(name)
if config_name == 'production':
    from config import ProductionConfig
    app.config.from_object(ProductionConfig)
else:
    from config import DevelopmentConfig
    app.config.from_object(DevelopmentConfig)

db = SQLAlchemy(app)
jwt = JWTManager(app)

# 注册Blueprint
from routes.auth import auth_bp
from routes.users import users_bp
app.register_blueprint(auth_bp)
app.register_blueprint(users_bp)

return app
undefined

Best Practices

最佳实践

✅ DO

✅ 建议

  • Use blueprints for modular organization
  • Implement proper authentication with JWT
  • Validate all user input
  • Use SQLAlchemy ORM for database operations
  • Implement comprehensive error handling
  • Use pagination for collection endpoints
  • Log errors and important events
  • Return appropriate HTTP status codes
  • Implement CORS properly
  • Use environment variables for configuration
  • 使用Blueprint实现模块化组织
  • 用JWT实现规范的认证机制
  • 验证所有用户输入
  • 使用SQLAlchemy ORM进行数据库操作
  • 实现全面的错误处理
  • 对集合端点使用分页
  • 记录错误和重要事件
  • 返回合适的HTTP状态码
  • 正确实现CORS
  • 使用环境变量管理配置

❌ DON'T

❌ 避免

  • Store secrets in code
  • Use global variables for shared state
  • Ignore database transactions
  • Trust user input without validation
  • Return stack traces in production
  • Use mutable default arguments
  • Forget to handle database connection errors
  • Implement authentication in route handlers
  • 在代码中存储密钥
  • 使用全局变量保存共享状态
  • 忽略数据库事务
  • 未验证就信任用户输入
  • 在生产环境返回堆栈跟踪
  • 使用可变默认参数
  • 忽略数据库连接错误处理
  • 在路由处理器中直接实现认证逻辑

Complete Example

完整示例

python
from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy
from flask_jwt_extended import JWTManager, create_access_token

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://user:pass@localhost/db'
db = SQLAlchemy(app)
jwt = JWTManager(app)

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String, unique=True)
    password = db.Column(db.String)

@app.route('/api/login', methods=['POST'])
def login():
    data = request.json
    user = User.query.filter_by(email=data['email']).first()
    if user:
        token = create_access_token(identity=user.id)
        return jsonify({'token': token}), 200
    return jsonify({'error': 'Invalid'}), 401

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

if __name__ == '__main__':
    app.run()
python
from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy
from flask_jwt_extended import JWTManager, create_access_token

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://user:pass@localhost/db'
db = SQLAlchemy(app)
jwt = JWTManager(app)

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String, unique=True)
    password = db.Column(db.String)

@app.route('/api/login', methods=['POST'])
def login():
    data = request.json
    user = User.query.filter_by(email=data['email']).first()
    if user:
        token = create_access_token(identity=user.id)
        return jsonify({'token': token}), 200
    return jsonify({'error': 'Invalid'}), 401

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

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