Flask测试平台后端开发实战
大约 7 分钟
Flask测试平台后端开发实战
上一篇我们聊了测试平台的整体架构，现在该撸起袖子干活了！今天我们要用Flask搭建一个稳定可靠的后端服务。别担心，我会用最接地气的方式教你，保证比看小说还有趣！
🎯 为什么选择Flask？
Flask vs Django：轻装上阵还是全副武装？
想象一下，你要去爬山：
- Django就像全副武装的登山队，装备齐全但负重太大
- Flask就像轻装简行的背包客，灵活机动想加什么装备随时加
对于测试平台开发，Flask的优势明显：
- 学习成本低：几行代码就能跑起来，成就感满满
- 扩展性强：需要什么功能就加什么，不强制绑定
- 社区活跃：遇到问题Google一下，答案一大堆
- 部署简单：不需要复杂的配置，开箱即用
🏗️ 项目结构设计
目录结构：井井有条才能走得远
flask_test_platform/
├── app/
│ ├── __init__.py # 应用工厂
│ ├── models/ # 数据模型
│ │ ├── __init__.py
│ │ ├── base.py # 基础模型
│ │ ├── project.py # 项目模型
│ │ ├── testcase.py # 测试用例模型
│ │ └── report.py # 报告模型
│ ├── api/ # API蓝图
│ │ ├── __init__.py
│ │ ├── project.py # 项目相关API
│ │ ├── testcase.py # 用例相关API
│ │ └── report.py # 报告相关API
│ ├── services/ # 业务逻辑层
│ │ ├── __init__.py
│ │ ├── project_service.py
│ │ ├── testcase_service.py
│ │ └── report_service.py
│ ├── utils/ # 工具函数
│ │ ├── __init__.py
│ │ ├── response.py # 统一响应格式
│ │ ├── validator.py # 参数验证
│ │ └── logger.py # 日志配置
│ └── config.py # 配置文件
├── migrations/ # 数据库迁移文件
├── tests/ # 单元测试
├── requirements.txt # 依赖包
├── run.py # 启动文件
└── README.md # 项目说明应用工厂模式：像搭积木一样组装应用
app/init.py：
from flask import Flask
from app.config import Config
from app.models import db
from app.api import register_blueprints
def create_app(config_class=Config):
"""应用工厂函数 - 像工厂一样生产Flask应用"""
app = Flask(__name__)
app.config.from_object(config_class)
# 初始化数据库
db.init_app(app)
# 注册蓝图
register_blueprints(app)
return app🗄️ 数据模型设计
Peewee ORM：轻量级的数据库操作神器
为什么选择Peewee而不是SQLAlchemy？
- 简单直观：语法接近Django ORM，学习成本低
- 轻量级：不会给项目增加太多负担
- 功能够用：对于测试平台来说完全够用
基础模型：所有模型的"祖宗"
app/models/base.py：
from peewee import *
from datetime import datetime
import os
# 数据库配置
DATABASE_URL = os.getenv('DATABASE_URL', 'mysql://user:password@localhost/test_platform')
db = MySQLDatabase(
'test_platform',
user='root',
password='123456',
host='localhost',
port=3306,
charset='utf8mb4'
)
class BaseModel(Model):
"""基础模型 - 所有模型的父类"""
id = AutoField(primary_key=True, verbose_name="主键ID")
created_at = DateTimeField(default=datetime.now, verbose_name="创建时间")
updated_at = DateTimeField(default=datetime.now, verbose_name="更新时间")
is_deleted = BooleanField(default=False, verbose_name="是否删除")
class Meta:
database = db
def save(self, *args, **kwargs):
"""重写保存方法，自动更新时间"""
self.updated_at = datetime.now()
return super().save(*args, **kwargs)
def soft_delete(self):
"""软删除 - 不真正删除数据，只标记为已删除"""
self.is_deleted = True
self.save()项目模型：测试项目的数字化身
app/models/project.py：
from peewee import *
from .base import BaseModel
class Project(BaseModel):
"""项目模型 - 测试项目的数据表示"""
name = CharField(max_length=100, unique=True, verbose_name="项目名称")
description = TextField(null=True, verbose_name="项目描述")
owner = CharField(max_length=50, verbose_name="项目负责人")
status = CharField(
max_length=20,
choices=[('active', '活跃'), ('inactive', '非活跃')],
default='active',
verbose_name="项目状态"
)
class Meta:
table_name = 'projects'
def __str__(self):
return f"Project({self.name})"
@classmethod
def get_active_projects(cls):
"""获取所有活跃项目"""
return cls.select().where(
(cls.status == 'active') &
(cls.is_deleted == False)
)测试用例模型：用例管理的核心
app/models/testcase.py：
from peewee import *
from .base import BaseModel
from .project import Project
import json
class TestCase(BaseModel):
"""测试用例模型"""
project = ForeignKeyField(Project, backref='testcases', verbose_name="所属项目")
name = CharField(max_length=200, verbose_name="用例名称")
type = CharField(
max_length=20,
choices=[('api', 'API测试'), ('ui', 'UI测试'), ('database', '数据库测试')],
verbose_name="用例类型"
)
priority = CharField(
max_length=10,
choices=[('high', '高'), ('medium', '中'), ('low', '低')],
default='medium',
verbose_name="优先级"
)
content = TextField(verbose_name="用例内容") # JSON格式存储
status = CharField(
max_length=20,
choices=[('active', '启用'), ('disabled', '禁用')],
default='active',
verbose_name="用例状态"
)
class Meta:
table_name = 'test_cases'
indexes = (
(('project', 'name'), True), # 项目内用例名称唯一
)
def get_content_dict(self):
"""获取用例内容的字典格式"""
try:
return json.loads(self.content)
except (json.JSONDecodeError, TypeError):
return {}
def set_content_dict(self, content_dict):
"""设置用例内容"""
self.content = json.dumps(content_dict, ensure_ascii=False)🔧 API接口设计
统一响应格式：让前端开发者爱上你的接口
app/utils/response.py：
from flask import jsonify
class APIResponse:
"""统一API响应格式"""
@staticmethod
def success(data=None, message="操作成功", code=200):
"""成功响应"""
return jsonify({
'success': True,
'code': code,
'message': message,
'data': data
}), code
@staticmethod
def error(message="操作失败", code=400, data=None):
"""错误响应"""
return jsonify({
'success': False,
'code': code,
'message': message,
'data': data
}), code
@staticmethod
def list_response(items, total, page=1, page_size=10):
"""列表响应 - 带分页信息"""
return jsonify({
'success': True,
'code': 200,
'message': "查询成功",
'data': {
'items': items,
'pagination': {
'total': total,
'page': page,
'page_size': page_size,
'total_pages': (total + page_size - 1) // page_size
}
}
})项目管理API：CRUD操作的标准实现
app/api/project.py：
from flask import Blueprint, request
from app.models.project import Project
from app.utils.response import APIResponse
from app.utils.validator import validate_required_fields
from playhouse.shortcuts import model_to_dict
from peewee import DoesNotExist
project_bp = Blueprint('project', __name__, url_prefix='/api/projects')
@project_bp.route('', methods=['GET'])
def get_projects():
"""获取项目列表"""
try:
page = int(request.args.get('page', 1))
page_size = int(request.args.get('page_size', 10))
# 查询项目
query = Project.get_active_projects()
total = query.count()
# 分页
projects = query.paginate(page, page_size)
# 转换为字典
project_list = [
model_to_dict(project, exclude=[Project.is_deleted])
for project in projects
]
return APIResponse.list_response(project_list, total, page, page_size)
except Exception as e:
return APIResponse.error(f"查询失败: {str(e)}")
@project_bp.route('', methods=['POST'])
def create_project():
"""创建项目"""
try:
data = request.get_json()
# 参数验证
required_fields = ['name', 'owner']
if not validate_required_fields(data, required_fields):
return APIResponse.error("缺少必要参数")
# 检查项目名是否已存在
if Project.select().where(Project.name == data['name']).exists():
return APIResponse.error("项目名称已存在")
# 创建项目
project = Project.create(
name=data['name'],
description=data.get('description', ''),
owner=data['owner']
)
return APIResponse.success(
data=model_to_dict(project, exclude=[Project.is_deleted]),
message="项目创建成功"
)
except Exception as e:
return APIResponse.error(f"创建失败: {str(e)}")
@project_bp.route('/<int:project_id>', methods=['GET'])
def get_project(project_id):
"""获取项目详情"""
try:
project = Project.get_by_id(project_id)
if project.is_deleted:
return APIResponse.error("项目不存在", 404)
return APIResponse.success(
data=model_to_dict(project, exclude=[Project.is_deleted])
)
except DoesNotExist:
return APIResponse.error("项目不存在", 404)
except Exception as e:
return APIResponse.error(f"查询失败: {str(e)}")
@project_bp.route('/<int:project_id>', methods=['PUT'])
def update_project(project_id):
"""更新项目"""
try:
data = request.get_json()
project = Project.get_by_id(project_id)
if project.is_deleted:
return APIResponse.error("项目不存在", 404)
# 更新字段
if 'name' in data:
project.name = data['name']
if 'description' in data:
project.description = data['description']
if 'owner' in data:
project.owner = data['owner']
if 'status' in data:
project.status = data['status']
project.save()
return APIResponse.success(
data=model_to_dict(project, exclude=[Project.is_deleted]),
message="项目更新成功"
)
except DoesNotExist:
return APIResponse.error("项目不存在", 404)
except Exception as e:
return APIResponse.error(f"更新失败: {str(e)}")
@project_bp.route('/<int:project_id>', methods=['DELETE'])
def delete_project(project_id):
"""删除项目（软删除）"""
try:
project = Project.get_by_id(project_id)
if project.is_deleted:
return APIResponse.error("项目不存在", 404)
project.soft_delete()
return APIResponse.success(message="项目删除成功")
except DoesNotExist:
return APIResponse.error("项目不存在", 404)
except Exception as e:
return APIResponse.error(f"删除失败: {str(e)}")🛠️ 业务逻辑层
服务层设计：让控制器更清爽
app/services/project_service.py：
from app.models.project import Project
from playhouse.shortcuts import model_to_dict
from peewee import DoesNotExist
class ProjectService:
"""项目服务类 - 封装项目相关的业务逻辑"""
@staticmethod
def get_project_list(page=1, page_size=10, filters=None):
"""获取项目列表"""
query = Project.get_active_projects()
# 应用过滤条件
if filters:
if filters.get('name'):
query = query.where(Project.name.contains(filters['name']))
if filters.get('owner'):
query = query.where(Project.owner == filters['owner'])
if filters.get('status'):
query = query.where(Project.status == filters['status'])
total = query.count()
projects = query.paginate(page, page_size)
return {
'items': [model_to_dict(p, exclude=[Project.is_deleted]) for p in projects],
'total': total,
'page': page,
'page_size': page_size
}
@staticmethod
def create_project(data):
"""创建项目"""
# 业务逻辑验证
if Project.select().where(Project.name == data['name']).exists():
raise ValueError("项目名称已存在")
project = Project.create(**data)
return model_to_dict(project, exclude=[Project.is_deleted])
@staticmethod
def get_project_stats(project_id):
"""获取项目统计信息"""
try:
project = Project.get_by_id(project_id)
if project.is_deleted:
raise DoesNotExist()
# 统计测试用例数量
total_cases = project.testcases.count()
active_cases = project.testcases.where(TestCase.status == 'active').count()
return {
'project_info': model_to_dict(project, exclude=[Project.is_deleted]),
'stats': {
'total_cases': total_cases,
'active_cases': active_cases,
'disabled_cases': total_cases - active_cases
}
}
except DoesNotExist:
raise ValueError("项目不存在")🔍 参数验证与错误处理
参数验证：防止垃圾数据进入系统
app/utils/validator.py：
from functools import wraps
from flask import request
from app.utils.response import APIResponse
def validate_required_fields(data, required_fields):
"""验证必需字段"""
if not data:
return False
for field in required_fields:
if field not in data or not data[field]:
return False
return True
def validate_json(required_fields=None):
"""装饰器：验证JSON参数"""
def decorator(f):
@wraps(f)
def decorated_function(*args, **kwargs):
if not request.is_json:
return APIResponse.error("请求必须是JSON格式")
data = request.get_json()
if not data:
return APIResponse.error("请求体不能为空")
if required_fields and not validate_required_fields(data, required_fields):
return APIResponse.error(f"缺少必要参数: {', '.join(required_fields)}")
return f(*args, **kwargs)
return decorated_function
return decorator🎯 下一步预告
今天我们搭建了一个稳定的Flask后端框架，实现了：
- 清晰的项目结构
- 灵活的数据模型
- 标准的API接口
- 完善的错误处理
下一篇我们将学习React前端开发，看看如何用现代化的前端技术打造用户友好的界面。记住，后端是大脑，前端是脸面，两者缺一不可！
💡 开发小贴士：写代码就像写文章，结构清晰、逻辑合理比炫技更重要。记住，代码是写给人看的，顺便让机器执行而已！