点击上方卡片关注我
设置星标 学习更多技能
在当今数字化时代，REST API已成为Web应用开发中不可或缺的一部分。它为不同系统之间的数据交互提供了标准化的接口，使得应用程序可以更加灵活、高效地进行通信。Python的Flask框架因其轻量级、灵活性和易于上手的特点，成为了快速搭建REST API的理想选择。
使用Flask-RESTful扩展
通过继承flask_restful.Resource类，可以将API的不同HTTP方法（GET、POST、PUT、DELETE）组织在同一个类中，使代码更加模块化。
from flask import Flask
from
flask_restful import Api, Resource, reqparse
app = Flask(__name__)
api = Api(app)
# 定义请求解析器
parser = reqparse.RequestParser()
parser.add_argument('name', type=str, help='Name is required', required=True)
parser.add_argument('age', type=int, help='Age must be an integer')
# 定义资源类
class UserResource(Resource):
def get(self, user_id=None):
if user_id:
# 返回单个用户
return {'id': user_id, 'name': 'John', 'age': 30}
else:
# 返回所有用户
return [{'id': 1, 'name': 'John', 'age': 30},
{'id': 2, 'name': 'Jane', 'age': 25}]
def post(self):
args = parser.parse_args()
# 创建新用户
return {'message': 'User created successfully', 'data': args}, 201
def put(self, user_id):
args = parser.parse_args()
# 更新用户信息
return {'message': 'User updated successfully', 'data': args}
def delete(self, user_id):
# 删除用户
return {'message': 'User deleted successfully'}, 204
# 添加资源到API
api.add_resource(UserResource, '/users', '/users/')
if __name__ == '__main__':
app.run(debug=True)
这个例子展示了如何使用Flask-RESTful创建一个处理用户资源的API。通过一个类处理所有HTTP方法，代码更加简洁和易于维护。
使用reqparse进行请求验证
Flask-RESTful的reqparse模块可以帮助我们验证和解析请求数据，确保输入的有效性。
from flask_restful import reqparse
# 创建请求解析器
parser = reqparse.RequestParser()
# 添加参数验证
parser.add_argument('name', type=str, required=True, help='Name cannot be blank')
parser.add_argument('email', type=str, required=True, help='Email is required')
parser.add_argument('age', type=int, choices=range(1, 100), help='Age must be between 1 and 99')
parser.add_argument('scores', type=list, location='json') # 处理JSON数组
class UserRegistration(Resource):
def post(self):
args = parser.parse_args()
# 处理验证后的数据
return {'message': 'Registration successful', 'data': args}
reqparse提供了多种验证选项，如类型检查、必填字段、范围限制等，可以有效防止无效数据进入系统。
处理不同格式的请求数据
Flask可以轻松处理多种格式的请求数据，如JSON、表单数据等。
from flask import request
class DataHandler(Resource):
def post(self):
# 处理JSON数据
if request.is_json:
data = request.get_json()
return {'message': 'JSON data received', 'data': data}
# 处理表单数据
elif request.form:
data = request.form.to_dict()
return {'message': 'Form data received', 'data': data}
# 处理文件上传
elif'file'in request.files:
file = request.files['file']
# 保存文件
file.save('uploaded_file.txt')
return {'message': 'File uploaded successfully'}
else:
return {'error': 'Unsupported media type'}, 415
这个例子展示了如何根据请求内容类型处理不同格式的数据，使API更加灵活。
统一API响应格式
为了使API响应更加一致和易于使用，建议定义统一的响应格式。
from flask import jsonify
def format_response(data=None, message=None, status=200):
response = {
'status': status,
'message': message,
'data': data
}
return jsonify(response), status
class ProductResource(Resource):
def get(self, product_id):
# 获取产品数据
product = {'id': product_id, 'name': 'iPhone', 'price': 999}
return format_response(product, 'Product retrieved successfully')
def post(self):
# 创建新产品
return format_response(None, 'Product created successfully', 201)
统一的响应格式可以提高API的可维护性和用户体验，使前端开发人员更容易处理返回的数据。
自定义错误处理
Flask可以自定义错误处理函数，为API提供统一的错误响应。
from flask import jsonify
@app.errorhandler(404)
def not_found(error):
return jsonify({
'status': 404,
'message': 'Resource not found'
}), 404
@app.errorhandler(500)
def internal_error(error):
return jsonify({
'status': 500,
'message': 'Internal server error. Please try again later.'
}), 500
# 处理Flask-RESTful的异常
from flask_restful import Api, abort
class CustomAPI(Api):
def handle_error(self, e):
# 处理自定义异常
if hasattr(e, 'status_code'):
return jsonify({
'status': e.status_code,
'message': str(e)
}), e.status_code
# 处理其他异常
return super().handle_error(e)
api = CustomAPI(app)
自定义错误处理可以确保API在遇到错误时返回一致的格式，提高用户体验。
使用JWT进行身份验证
JSON Web Token (JWT) 是一种轻量级的身份验证机制，非常适合REST API。
from flask import request, jsonify
from flask_restful import Resource
import jwt
from datetime import datetime, timedelta
from functools import wraps
# 密钥，用于签名JWT
SECRET_KEY = 'your-secret-key'
# 生成JWT
def generate_token(user_id):
payload = {
'user_id': user_id,
'exp': datetime.utcnow() + timedelta(hours=1)
}
return jwt.encode(payload, SECRET_KEY, algorithm='HS256')
# 验证JWT
def token_required(f):
@wraps(f)
def decorated(*args, **kwargs):
token = None
# 从请求头中获取token
if'Authorization'in request.headers:
token = request.headers['Authorization'].split(' ')[1]
ifnot token:
return jsonify({'message': 'Token is missing!'}), 401
try:
# 验证token
data = jwt.decode(token, SECRET_KEY, algorithms=['HS256'])
current_user = data['user_id']
except:
return jsonify({'message': 'Invalid token!'}), 401
return f(current_user, *args, **kwargs)
return decorated
# 登录接口
class Login(Resource):
def post(self):
# 验证用户 credentials
# 这里简化处理，实际应用中应该查询数据库
user_id = 1# 假设验证通过
token = generate_token(user_id)
return jsonify({'token': token})
# 需要认证的受保护资源
class ProtectedResource(Resource):
@token_required
def get(self, user_id):
return jsonify({'message': 'This is a protected resource', 'user_id': user_id})
JWT提供了无状态的身份验证机制，非常适合微服务和分布式系统。
使用SQLAlchemy进行数据库操作
Flask-SQLAlchemy是Flask的一个扩展，简化了与SQL数据库的交互。
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_restful import Api, Resource
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///example.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)
api = Api(app)
# 定义数据模型
class User(db.Model):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(80))
email = db.Column(db.String(120), unique=
True)
def to_dict(self):
return {
'id': self.id,
'name': self.name,
'email': self.email
}
# 创建所有表
with app.app_context():
db.create_all()
# 资源类
class UserListResource(Resource):
def get(self):
users = User.query.all()
return [user.to_dict() for user in users]
def post(self):
data = request.get_json()
new_user = User(name=data['name'], email=data['email'])
db.session.add(new_user)
db.session.commit()
return new_user.to_dict(), 201
class UserResource(Resource):
def get(self, user_id):
user = User.query.get_or_404(user_id)
return user.to_dict()
def put(self, 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 user.to_dict()
def delete(self, user_id):
user = User.query.get_or_404(user_id)
db.session.delete(user)
db.session.commit()
return'', 204
# 添加资源到API
api.add_resource(UserListResource, '/users')
api.add_resource(UserResource, '/users/')
SQLAlchemy提供了ORM（对象关系映射）功能，使我们可以使用Python类和对象来操作数据库，而不必编写SQL语句。
实现请求限流保护API
为了防止API被滥用，可以实现请求限流机制。
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
app = Flask(__name__)
api = Api(app)
# 初始化Limiter
limiter = Limiter(
app,
key_func=get_remote_address,
default_limits=["200 per day", "50 per hour"]
)
# 应用限流到特定资源
class ExpensiveResource(Resource):
decorators = [limiter.limit("10 per minute")]
def get(self):
# 执行耗时操作
return {"message": "This is an expensive operation"}
# 对所有API应用默认限流
api.add_resource(ExpensiveResource, '/expensive')
请求限流可以保护API免受恶意攻击和过度使用，确保服务的稳定性。
自动生成API文档
使用Flask-RESTful和Flasgger可以自动生成交互式API文档。
from flasgger import Swagger
app = Flask(__name__)
api = Api(app)
swagger = Swagger(app)
class UserResource(Resource):
def get(self, user_id):
"""
获取用户信息
---
tags:
- 用户
parameters:
- name: user_id
in: path
type: integer
required: true
description: 用户ID
responses:
200:
description: 用户信息
schema:
type: object
properties:
id:
type: integer
name:
type: string
email:
type: string
404:
description: 用户不存在
"""
user = User.query.get_or_404(user_id)
return user.to_dict()
api.add_resource(UserResource, '/users/')
访问/apidocs路径，你将看到一个交互式的API文档界面，用户可以在其中测试API端点。
总结
使用 Flask 搭建 REST API 时，掌握核心技巧能大幅提升开发效率。这 10 个技巧覆盖从开发到上线的全流程，帮助大家可以快速构建专业、健壮的 REST API 服务。
如果你觉得文章还不错，请大家 点赞、分享、留言 下，因为这将是我持续输出更多优质文章的最强动力！
我们还为大家准备了Python资料，感兴趣的小伙伴快来找我领取一起交流学习哦！
