387 lines
12 KiB
Python
387 lines
12 KiB
Python
"""
|
||
演示：使用Service层和中间件优化API
|
||
|
||
这个文件展示了如何使用Service层和中间件来重构和优化API
|
||
"""
|
||
|
||
from flask import request, jsonify, current_app
|
||
from app import db
|
||
from app.api import api_bp
|
||
from app.services.license_service import LicenseService
|
||
from app.services.product_service import ProductService
|
||
from app.middleware.rate_limit import rate_limit, user_key, ip_key
|
||
from app.utils.file_security import secure_file_upload
|
||
from .decorators import require_login, require_admin
|
||
import os
|
||
|
||
|
||
# ==================== 优化后的License API ====================
|
||
|
||
@api_bp.route('/licenses', methods=['GET'])
|
||
@require_login
|
||
@rate_limit(limit=100, window=3600, key_func=user_key) # 用户每小时100次请求
|
||
def get_licenses_optimized():
|
||
"""
|
||
获取卡密列表（优化版）
|
||
使用Service层解耦业务逻辑
|
||
"""
|
||
try:
|
||
# 获取查询参数
|
||
page = request.args.get('page', 1, type=int)
|
||
per_page = min(request.args.get('per_page', 20, type=int), 100)
|
||
keyword = request.args.get('keyword', '').strip()
|
||
license_type = request.args.get('type', type=int)
|
||
status = request.args.get('status', type=int)
|
||
product_id = request.args.get('product_id')
|
||
|
||
# 使用Service层获取数据
|
||
licenses, total = LicenseService.get_licenses(
|
||
page=page,
|
||
per_page=per_page,
|
||
keyword=keyword,
|
||
license_type=license_type,
|
||
status=status,
|
||
product_id=product_id
|
||
)
|
||
|
||
return jsonify({
|
||
'success': True,
|
||
'data': {
|
||
'licenses': [license.to_dict() for license in licenses],
|
||
'pagination': {
|
||
'page': page,
|
||
'per_page': per_page,
|
||
'total': total,
|
||
'pages': (total + per_page - 1) // per_page
|
||
}
|
||
}
|
||
}), 200
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"获取卡密列表失败: {str(e)}")
|
||
return jsonify({
|
||
'success': False,
|
||
'message': '服务器内部错误，请稍后重试'
|
||
}), 500
|
||
|
||
|
||
@api_bp.route('/licenses/generate', methods=['POST'])
|
||
@require_admin
|
||
@rate_limit(limit=10, window=3600, key_func=user_key) # 超级管理员每小时10次生成请求
|
||
def generate_license_optimized():
|
||
"""
|
||
生成卡密（优化版）
|
||
使用Service层处理业务逻辑
|
||
"""
|
||
try:
|
||
data = request.get_json()
|
||
if not data:
|
||
return jsonify({
|
||
'success': False,
|
||
'message': '请提供JSON数据'
|
||
}), 400
|
||
|
||
# 验证必填参数
|
||
required_fields = ['product_id', 'valid_days', 'count']
|
||
for field in required_fields:
|
||
if field not in data:
|
||
return jsonify({
|
||
'success': False,
|
||
'message': f'缺少必填参数: {field}'
|
||
}), 400
|
||
|
||
product_id = data['product_id']
|
||
valid_days = data['valid_days']
|
||
count = data['count']
|
||
license_type = data.get('type', 0)
|
||
max_bind_times = data.get('max_bind_times', 1)
|
||
remark = data.get('remark', '')
|
||
|
||
# 验证参数
|
||
if valid_days <= 0:
|
||
return jsonify({
|
||
'success': False,
|
||
'message': '有效天数必须大于0'
|
||
}), 400
|
||
|
||
if count <= 0 or count > 1000:
|
||
return jsonify({
|
||
'success': False,
|
||
'message': '生成数量必须在1-1000之间'
|
||
}), 400
|
||
|
||
# 使用Service层创建卡密
|
||
success, msg, licenses = LicenseService.batch_create_licenses(
|
||
product_id=product_id,
|
||
valid_days=valid_days,
|
||
count=count,
|
||
license_type=license_type,
|
||
max_bind_times=max_bind_times,
|
||
remark=remark
|
||
)
|
||
|
||
if not success:
|
||
return jsonify({
|
||
'success': False,
|
||
'message': msg
|
||
}), 400
|
||
|
||
return jsonify({
|
||
'success': True,
|
||
'message': msg,
|
||
'data': {
|
||
'licenses': [license.to_dict() for license in licenses]
|
||
}
|
||
}), 201
|
||
|
||
except Exception as e:
|
||
db.session.rollback()
|
||
current_app.logger.error(f"生成卡密失败: {str(e)}")
|
||
return jsonify({
|
||
'success': False,
|
||
'message': f'服务器内部错误: {str(e)}'
|
||
}), 500
|
||
|
||
|
||
@api_bp.route('/licenses/verify', methods=['GET'])
|
||
@rate_limit(limit=1000, window=3600, key_func=ip_key) # 客户端每小时1000次验证请求
|
||
def verify_license_optimized():
|
||
"""
|
||
验证卡密（优化版）
|
||
使用Service层处理业务逻辑，添加频率限制
|
||
"""
|
||
try:
|
||
license_key = request.args.get('license_key')
|
||
machine_code = request.args.get('machine_code')
|
||
software_version = request.args.get('software_version', '')
|
||
|
||
if not license_key or not machine_code:
|
||
return jsonify({
|
||
'success': False,
|
||
'message': '缺少必填参数: license_key, machine_code'
|
||
}), 400
|
||
|
||
# 使用Service层验证卡密
|
||
success, msg, license = LicenseService.verify_license(
|
||
license_key=license_key,
|
||
machine_code=machine_code,
|
||
software_version=software_version
|
||
)
|
||
|
||
if not success:
|
||
return jsonify({
|
||
'success': False,
|
||
'message': msg
|
||
}), 400
|
||
|
||
return jsonify({
|
||
'success': True,
|
||
'message': msg,
|
||
'data': {
|
||
'license': license.to_dict()
|
||
}
|
||
}), 200
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"验证卡密失败: {str(e)}")
|
||
return jsonify({
|
||
'success': False,
|
||
'message': '服务器内部错误，请稍后重试'
|
||
}), 500
|
||
|
||
|
||
@api_bp.route('/licenses/unbind', methods=['POST'])
|
||
@require_login
|
||
@rate_limit(limit=50, window=3600, key_func=user_key) # 用户每小时50次解绑请求
|
||
def unbind_license_optimized():
|
||
"""
|
||
解绑卡密（优化版）
|
||
使用Service层处理业务逻辑
|
||
"""
|
||
try:
|
||
data = request.get_json()
|
||
if not data:
|
||
return jsonify({
|
||
'success': False,
|
||
'message': '请提供JSON数据'
|
||
}), 400
|
||
|
||
license_key = data.get('license_key')
|
||
if not license_key:
|
||
return jsonify({
|
||
'success': False,
|
||
'message': '缺少必填参数: license_key'
|
||
}), 400
|
||
|
||
# 使用Service层解绑卡密
|
||
success, msg = LicenseService.unbind_license(license_key)
|
||
|
||
if not success:
|
||
return jsonify({
|
||
'success': False,
|
||
'message': msg
|
||
}), 400
|
||
|
||
return jsonify({
|
||
'success': True,
|
||
'message': msg
|
||
}), 200
|
||
|
||
except Exception as e:
|
||
db.session.rollback()
|
||
current_app.logger.error(f"解绑卡密失败: {str(e)}")
|
||
return jsonify({
|
||
'success': False,
|
||
'message': '服务器内部错误，请稍后重试'
|
||
}), 500
|
||
|
||
|
||
# ==================== 优化后的产品API ====================
|
||
|
||
@api_bp.route('/products', methods=['GET'])
|
||
@rate_limit(limit=200, window=3600, key_func=ip_key) # 客户端每小时200次请求
|
||
def get_products_optimized():
|
||
"""
|
||
获取产品列表（优化版）
|
||
使用Service层，添加频率限制
|
||
"""
|
||
try:
|
||
page = request.args.get('page', 1, type=int)
|
||
per_page = min(request.args.get('per_page', 20, type=int), 100)
|
||
keyword = request.args.get('keyword', '').strip()
|
||
product_type = request.args.get('type')
|
||
is_paid = request.args.get('is_paid', type=int)
|
||
|
||
# 使用Service层获取产品
|
||
products, total = ProductService.get_products(
|
||
page=page,
|
||
per_page=per_page,
|
||
keyword=keyword,
|
||
product_type=product_type,
|
||
is_paid=is_paid
|
||
)
|
||
|
||
# 为每个产品添加最新版本信息
|
||
products_with_version = []
|
||
for product in products:
|
||
product_dict = product.to_dict()
|
||
latest_version = ProductService.get_latest_version(product.product_id)
|
||
product_dict['latest_version'] = latest_version.version_num if latest_version else None
|
||
products_with_version.append(product_dict)
|
||
|
||
return jsonify({
|
||
'success': True,
|
||
'data': {
|
||
'products': products_with_version,
|
||
'pagination': {
|
||
'page': page,
|
||
'per_page': per_page,
|
||
'total': total,
|
||
'pages': (total + per_page - 1) // per_page
|
||
}
|
||
}
|
||
}), 200
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"获取产品列表失败: {str(e)}")
|
||
return jsonify({
|
||
'success': False,
|
||
'message': '服务器内部错误，请稍后重试'
|
||
}), 500
|
||
|
||
|
||
# ==================== 优化后的文件上传API ====================
|
||
|
||
@api_bp.route('/upload', methods=['POST'])
|
||
@require_login
|
||
@rate_limit(limit=20, window=3600, key_func=user_key) # 用户每小时20次上传请求
|
||
def upload_file_optimized():
|
||
"""
|
||
文件上传（优化版）
|
||
使用安全的文件上传工具
|
||
"""
|
||
try:
|
||
if 'file' not in request.files:
|
||
return jsonify({
|
||
'success': False,
|
||
'message': '没有选择文件'
|
||
}), 400
|
||
|
||
file = request.files['file']
|
||
if file.filename == '':
|
||
return jsonify({
|
||
'success': False,
|
||
'message': '没有选择文件'
|
||
}), 400
|
||
|
||
# 获取上传目录
|
||
upload_folder = current_app.config.get('UPLOAD_FOLDER', 'static/uploads')
|
||
|
||
# 定义允许的MIME类型
|
||
allowed_mimetypes = {
|
||
'image/png', 'image/jpeg', 'image/gif', 'image/webp',
|
||
'application/pdf', 'text/plain',
|
||
'application/zip', 'application/x-zip-compressed'
|
||
}
|
||
|
||
# 使用安全的文件上传
|
||
success, msg, file_path = secure_file_upload(file, upload_folder, allowed_mimetypes)
|
||
|
||
if not success:
|
||
return jsonify({
|
||
'success': False,
|
||
'message': msg
|
||
}), 400
|
||
|
||
# 返回相对路径
|
||
relative_path = os.path.relpath(file_path, current_app.root_path)
|
||
|
||
return jsonify({
|
||
'success': True,
|
||
'message': msg,
|
||
'data': {
|
||
'file_path': relative_path,
|
||
'filename': file.filename,
|
||
'size': os.path.getsize(file_path)
|
||
}
|
||
}), 201
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"文件上传失败: {str(e)}")
|
||
return jsonify({
|
||
'success': False,
|
||
'message': f'文件上传失败: {str(e)}'
|
||
}), 500
|
||
|
||
|
||
# ==================== 使用示例 ====================
|
||
|
||
"""
|
||
总结：优化后的API具有以下优势：
|
||
|
||
1. **解耦业务逻辑**：
|
||
- API层只处理HTTP相关逻辑（请求解析、响应格式化）
|
||
- 业务逻辑转移到Service层
|
||
- 模型层只负责数据访问
|
||
|
||
2. **安全性增强**：
|
||
- 频率限制防止API滥用
|
||
- 参数验证确保数据安全
|
||
- 安全文件上传防止恶意文件
|
||
|
||
3. **可维护性提升**：
|
||
- 职责分离，每个组件职责单一
|
||
- 代码复用，Service层可以被多个API调用
|
||
- 易于测试，可以单独测试Service层
|
||
|
||
4. **性能优化**：
|
||
- 避免N+1查询问题
|
||
- 缓存常用数据
|
||
- 数据库索引优化
|
||
|
||
5. **监控和可观测性**：
|
||
- 健康检查端点
|
||
- 性能指标收集
|
||
- 日志记录
|
||
"""
|