RESTful API¶
📖 章节简介¶
本章将介绍如何使用Flask构建RESTful API，学习API设计原则、JSON响应处理和错误处理。
🌐 RESTful设计¶
1. RESTful原则¶
Python
# RESTful API设计原则
restful_principles = {
'资源导向': '一切皆资源，使用名词而非动词',
'统一接口': '使用标准的HTTP方法',
'无状态': '每个请求包含所有必要信息',
'可缓存': '响应应该明确是否可缓存',
'分层系统': '客户端不知道是否连接到终端服务器'
}
# HTTP方法与CRUD操作
http_methods = {
'GET': '读取资源',
'POST': '创建资源',
'PUT': '更新资源（完整）',
'PATCH': '更新资源（部分）',
'DELETE': '删除资源'
}
# 状态码
status_codes = {
'200': 'OK - 请求成功',
'201': 'Created - 资源创建成功',
'204': 'No Content - 成功但无返回内容',
'400': 'Bad Request - 请求参数错误',
'401': 'Unauthorized - 未授权',
'403': 'Forbidden - 禁止访问',
'404': 'Not Found - 资源不存在',
'500': 'Internal Server Error - 服务器错误'
}
2. API设计示例¶
Python
# RESTful API路由设计
api_routes = {
'用户资源': {
'GET /api/users': '获取用户列表',
'GET /api/users/<id>': '获取单个用户',
'POST /api/users': '创建用户',
'PUT /api/users/<id>': '更新用户',
'DELETE /api/users/<id>': '删除用户'
},
'文章资源': {
'GET /api/posts': '获取文章列表',
'GET /api/posts/<id>': '获取单篇文章',
'POST /api/posts': '创建文章',
'PUT /api/posts/<id>': '更新文章',
'DELETE /api/posts/<id>': '删除文章'
}
}
🔧 API实现¶
1. 基础API¶
Python
# app/api/routes.py
from flask import jsonify, request
from app.models import User, Post
@api_bp.route('/users', methods=['GET'])
def get_users():
"""获取用户列表"""
page = request.args.get('page', 1, type=int)
per_page = request.args.get('per_page', 10, type=int)
pagination = User.query.paginate(
page=page, per_page=per_page, error_out=False
)
users = [user.to_dict() for user in pagination.items]
return jsonify({
'users': users,
'total': pagination.total,
'pages': pagination.pages,
'current_page': page
})
@api_bp.route('/users/<int:user_id>', methods=['GET'])
def get_user(user_id):
"""获取单个用户"""
user = db.get_or_404(User, user_id)
return jsonify(user.to_dict())
@api_bp.route('/users', methods=['POST'])
def create_user():
"""创建用户"""
data = request.get_json()
if not data or 'username' not in data or 'email' not in data:
return jsonify({'error': '缺少必要字段'}), 400
user = User(username=data['username'], email=data['email'])
if 'password' in data:
user.set_password(data['password'])
db.session.add(user)
db.session.commit()
return jsonify(user.to_dict()), 201
@api_bp.route('/users/<int:user_id>', methods=['PUT'])
def update_user(user_id):
"""更新用户"""
user = db.get_or_404(User, user_id)
data = request.get_json()
for field in ['username', 'email']:
if field in data:
setattr(user, field, data[field])
if 'password' in data:
user.set_password(data['password'])
db.session.commit()
return jsonify(user.to_dict())
@api_bp.route('/users/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
"""删除用户"""
user = db.get_or_404(User, user_id)
db.session.delete(user)
db.session.commit()
return '', 204
2. 分页和过滤¶
Python
@api_bp.route('/posts', methods=['GET'])
def get_posts():
"""获取文章列表（支持分页和过滤）"""
# 获取查询参数
page = request.args.get('page', 1, type=int)
per_page = request.args.get('per_page', 10, type=int)
category = request.args.get('category')
author_id = request.args.get('author_id', type=int)
search = request.args.get('search')
# 构建查询
query = Post.query
# 过滤
if category:
query = query.join(Category).filter(Category.name == category)
if author_id:
query = query.filter(Post.user_id == author_id)
if search:
query = query.filter(
Post.title.contains(search) | Post.content.contains(search)
)
# 排序
sort = request.args.get('sort', 'created_at')
order = request.args.get('order', 'desc')
# 安全验证：sort参数白名单验证，防止任意属性访问漏洞
ALLOWED_SORT_FIELDS = {'created_at', 'updated_at', 'title', 'view_count', 'like_count'}
if sort not in ALLOWED_SORT_FIELDS:
sort = 'created_at' # 不合法时使用默认值
# order参数白名单验证
if order not in {'asc', 'desc'}:
order = 'desc' # 不合法时使用默认值
# 使用验证后的参数进行排序
sort_field = getattr(Post, sort)
if order == 'asc':
query = query.order_by(sort_field.asc())
else:
query = query.order_by(sort_field.desc())
# 分页
pagination = query.paginate(
page=page, per_page=per_page, error_out=False
)
posts = [post.to_dict() for post in pagination.items]
return jsonify({
'posts': posts,
'total': pagination.total,
'pages': pagination.pages,
'current_page': page,
'per_page': per_page
})
🛡️ 认证和授权¶
1. API认证¶
Python
# app/api/auth.py
from functools import wraps
from flask import request, jsonify
from flask_httpauth import HTTPTokenAuth
auth = HTTPTokenAuth(scheme='ApiKey', header='X-API-Key')
@auth.verify_token
def verify_token(token):
"""验证API密钥"""
if token:
user = User.query.filter_by(api_key=token).first()
if user and user.is_active:
return user
return None
@auth.error_handler
def unauthorized():
"""未授权响应"""
return jsonify({'error': '未授权'}), 401
# 装饰器
def api_required(f):
"""API认证装饰器"""
@wraps(f)
# *args和**kwargs透传所有参数给原函数f，使装饰器对任意签名的视图函数通用
def decorated(*args, **kwargs): # *args接收任意位置参数；**kwargs接收任意关键字参数
if not auth.current_user():
return unauthorized()
return f(*args, **kwargs)
return decorated
def admin_required(f):
"""管理员权限装饰器"""
@wraps(f)
def decorated(*args, **kwargs):
if not auth.current_user():
return unauthorized()
if not auth.current_user().is_admin:
return jsonify({'error': '权限不足'}), 403
return f(*args, **kwargs)
return decorated
2. 使用认证¶
Python
@api_bp.route('/admin/posts', methods=['POST'])
@admin_required
def create_post():
"""创建文章（需要管理员权限）"""
data = request.get_json()
post = Post(
title=data['title'],
content=data['content'],
author=auth.current_user()
)
db.session.add(post)
db.session.commit()
return jsonify(post.to_dict()), 201
⚠️ 错误处理¶
1. 统一错误处理¶
Python
# app/api/errors.py
from flask import jsonify
class APIError(Exception):
"""API错误基类"""
def __init__(self, message, status_code=400, payload=None):
super().__init__() # super()调用父类方法
self.message = message
self.status_code = status_code
self.payload = payload
def to_dict(self):
rv = dict(self.payload or ())
rv['message'] = self.message
return rv
class ValidationError(APIError):
"""验证错误"""
pass
class NotFoundError(APIError):
"""未找到错误"""
status_code = 404
class UnauthorizedError(APIError):
"""未授权错误"""
status_code = 401
class ForbiddenError(APIError):
"""禁止访问错误"""
status_code = 403
# 错误处理器
@api_bp.errorhandler(APIError)
def handle_api_error(error):
"""处理API错误"""
response = jsonify(error.to_dict())
response.status_code = error.status_code
return response
@api_bp.errorhandler(404)
def resource_not_found(error):
"""处理404错误"""
return jsonify({'error': '资源不存在'}), 404
@api_bp.errorhandler(500)
def internal_error(error):
"""处理500错误"""
db.session.rollback()
return jsonify({'error': '服务器内部错误'}), 500
2. 使用错误¶
Python
@api_bp.route('/users/<int:user_id>', methods=['GET'])
def get_user(user_id):
"""获取用户"""
user = db.session.get(User, user_id)
if not user:
raise NotFoundError('用户不存在')
return jsonify(user.to_dict())
@api_bp.route('/users', methods=['POST'])
def create_user():
"""创建用户"""
data = request.get_json()
# 验证
if not data:
raise ValidationError('请求数据不能为空')
if 'username' not in data:
raise ValidationError('用户名不能为空')
if 'email' not in data:
raise ValidationError('邮箱不能为空')
# 检查用户名是否已存在
if User.query.filter_by(username=data['username']).first():
raise ValidationError('用户名已被使用')
# 创建用户
user = User(username=data['username'], email=data['email'])
db.session.add(user)
db.session.commit()
return jsonify(user.to_dict()), 201
📚 API文档¶
1. Swagger集成¶
Python
# app/__init__.py
from flasgger import Swagger
def create_app():
app = Flask(__name__)
# 配置Swagger
swagger_config = {
"headers": [],
"specs": [
{
"endpoint": 'apispec',
"route": '/apispec.json',
"rule_filter": lambda rule: True, # lambda匿名函数：简洁的单行函数
"model_filter": lambda tag: True,
}
],
"static_url_path": "/flasgger_static",
"swagger_ui": True,
"specs_route": "/api/docs"
}
swagger = Swagger(app, config=swagger_config)
return app
2. API文档示例¶
Python
@api_bp.route('/users', methods=['POST'])
@swag_from({
'tags': ['用户'],
'parameters': [
{
'name': 'body',
'in': 'body',
'required': True,
'schema': {
'type': 'object',
'properties': {
'username': {'type': 'string', 'required': True},
'email': {'type': 'string', 'required': True},
'password': {'type': 'string', 'required': True}
}
}
}
],
'responses': {
'201': {
'description': '用户创建成功',
'schema': {
'type': 'object',
'properties': {
'id': {'type': 'integer'},
'username': {'type': 'string'},
'email': {'type': 'string'}
}
}
},
'400': {
'description': '请求参数错误'
}
}
})
def create_user():
"""创建用户"""
# 实现代码
pass
💡 最佳实践¶
1. API设计¶
Python
# API设计最佳实践
api_design = {
'版本控制': '使用版本号（/api/v1/users）',
'一致性': '保持API接口一致',
'错误处理': '返回统一的错误格式',
'限流': '实现API限流',
'文档': '提供完整的API文档'
}
2. 性能优化¶
Python
# API性能优化
api_performance = {
'缓存': '使用缓存减少数据库查询',
'分页': '实现分页避免大量数据传输',
'压缩': '启用响应压缩',
'异步': '使用异步处理提高性能',
'索引': '为查询字段创建索引'
}
📝 练习题¶
基础题¶
- 什么是RESTful API？
- HTTP方法有哪些？分别对应什么操作？
- 如何设计RESTful API路由？
进阶题¶
- 实现API认证。
- 处理API错误。
- 编写API文档。
实践题¶
- 创建完整的用户API。
- 实现文章CRUD API。
- 构建API限流功能。
📚 推荐阅读¶
🔗 下一章¶
部署与运维 - 学习Flask应用的部署和运维。