日常多表查询时，容易引入脏数据，需要删除多表中脏数据，减少直接操作DB的工作量
假设有表如下：
TestTaskPackagePluginApplianceRel 中间表，挂钩插件包和插件设备
TestTaskPackage 插件包
TestTaskPluginAppliance 插件设备
python3from django.db import transaction class TestTaskPackage(UUIDModel, DatedModel): ... def delete_with_related(self): """删除当前 TestTaskPackage 及其关联的 TestTaskPluginAppliance 和中间表记录""" with transaction.atomic(): # 先查出来关联TestTaskPluginAppliance的记录并强制评估为列表 appliance_ids = TestTaskPackagePluginApplianceRel.objects.filter(test_task_package=self).values_list( 'test_task_plugin_appliance_id', flat=True ) appliances = list(TestTaskPluginAppliance.objects.filter(id__in=appliance_ids)) logger.info(f'appliances: {appliances}') # 删除中间表中的记录 rels = TestTaskPackagePluginApplianceRel.objects.filter(test_task_package=self) for rel in rels: logger.info(f"Deleting relation: {rel}") rel.delete() # 删除关联的 TestTaskPluginAppliance # logger.info(f'appliances after: {appliances}') for appliance in appliances: logger.info(f"Deleting appliance: {appliance}") appliance.delete() # 删除当前 TestTaskPackage logger.info(f"Deleting TestTaskPackage: {self}") self.delete()
python3with transaction.atomic():
作用：把代码块中的数据库操作放在同一个事务里执行。
好处：
如果中间任何一步抛异常，所有已执行的 SQL 都会回滚，保证数据一致性。
避免出现删除了一部分关联记录但没有完全清理干净的情况
python3appliance_ids = TestTaskPackagePluginApplianceRel.objects.filter( test_task_package=self ).values_list('test_task_plugin_appliance_id', flat=True)
filter：筛选出中间表 TestTaskPackagePluginApplianceRel 中，关联到当前 TestTaskPackage 的记录。
values_list(..., flat=True)：直接取出字段值列表（只要一个字段时 flat=True 才能得到一维列表）。
这里的用途：先获取所有相关的 TestTaskPluginAppliance ID，方便后面批量删除
python3appliances = list(TestTaskPluginAppliance.objects.filter(id__in=appliance_ids))
jango 的 QuerySet 是惰性执行的（lazy evaluation）。
用 list(...) 会立刻执行 SQL 查询，把结果加载到内存中。
这样做的原因：
后续删除中间表记录时，这些对象仍然能在内存中访问，避免因级联删除或 QuerySet 变化导致数据不一致。
python3class TestTaskPackageViewSet(viewsets.ModelViewSet): queryset = m.TestTaskPackage.objects.all() serializer_class = s.TestTaskPackageSerializer filter_fields = '__all__' ordering = ['-pk'] permission_classes = [ActionBasedPermission] action_permissions = { 'list,retrieve': IsMember, 'create': IsMember, 'update,partial_update,destroy,delete_with_related': IsMember | UserWithCondition.build(lambda u, o: u == o.author), } @action(methods=['POST'], detail=False) def delete_with_related(self, request): """删除指定的 TestTaskPackage 及其关联的记录""" from rest_framework import status try: test_task_id = request.data.get('test_task_id') logger.info(f'test_task_id: {test_task_id}') test_task_package = (m.TestTaskPackage.objects.filter(test_task_id=test_task_id, package_type=m.TestTaskPackage.TYPE_PLUGIN_PACKAGE). order_by('date_created').first()) logger.info(f'test_task_package: {test_task_package}') # 检查 test_task_package 是否为空 if not test_task_package: return Response( {'status': 'info', 'message': '当前没有关联任何插件包，跳过删除操作。'}, status=status.HTTP_200_OK ) test_task_package.delete_with_related() return Response( {'status': 'success', 'message': 'TestTaskPackage and related records deleted successfully.'}, status=status.HTTP_204_NO_CONTENT) except Exception as e: return Response({'status': 'error', 'message': str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
python3from XXX.models import GrpcMemberProxy from django.test import TestCase from plugin_test import models as m import uuid from datetime import datetime class TestTaskPackageModelTest(TestCase): def setUp(self): # 创建一个 TestTaskPackage 实例 author = GrpcMemberProxy(username='abc') self.test_task_package = m.TestTaskPackage.objects.create( id=uuid.uuid4(), test_task_id=uuid.uuid4(), mobile_device_id=123456789, file_md5='dummy_md5', author=author ) # 创建一个 TestTaskPluginAppliance 实例 self.test_task_plugin_appliance = m.TestTaskPluginAppliance.objects.create( name='Test Appliance', appliance_code='dummy_code', sn='dummy_sn', bt_mac='dummy_bt_mac', enterprise_name='Dummy Enterprise', appliance_type='Dummy Type', enterprise_code='dummy_ent_code', cat_code='dummy_cat_code', sn8='dummy_sn8', develop_group_name='Dummy Group', product_code='dummy_product_code', wifi_version='dummy_wifi_version', register_time=datetime.now(), # 提供当前时间作为注册时间 a0='dummy_a0', prd_name='Dummy Product', prd_id='dummy_prd_id', model_type='dummy_model_type', access_method='dummy_access_method', source='dummy_source', product_code_in_app_svc='dummy_product_code_in_app_svc', smart_product_id=1, prd_info={}, # 提供一个空字典作为默认值 prd_id_on_spid='dummy_prd_id_on_spid' ) # 创建一个中间表记录 m.TestTaskPackagePluginApplianceRel.objects.create( test_task_package=self.test_task_package, test_task_plugin_appliance=self.test_task_plugin_appliance ) def test_delete_with_related(self): # 确保中间表记录存在 self.assertEqual(m.TestTaskPackagePluginApplianceRel.objects.count(), 1) # 确保 TestTaskPluginAppliance 记录存在 self.assertEqual(m.TestTaskPluginAppliance.objects.count(), 1) # 调用 delete_with_related 方法 self.test_task_package.delete_with_related() # 检查所有相关记录是否被删除 self.assertEqual(m.TestTaskPackage.objects.filter(id=self.test_task_package.id).count(), 0) self.assertEqual(m.TestTaskPluginAppliance.objects.filter(id=self.test_task_plugin_appliance.id).count(), 0) self.assertEqual(m.TestTaskPackagePluginApplianceRel.objects.count(), 0)
python3class TestTaskPackageModelTest(TestCase):
继承自 django.test.TestCase，意味着：
每个 test_ 方法运行前都会新建一个测试数据库（事务隔离）。
每个测试方法运行后会回滚数据库，不影响其他测试。
运行速度快且测试之间数据互不干扰。
python3def setUp(self):
setUp 会在每个测试方法执行前自动运行，用来准备测试数据。
python3m.TestTaskPackagePluginApplianceRel.objects.create( test_task_package=self.test_task_package, test_task_plugin_appliance=self.test_task_plugin_appliance )
建立 TestTaskPackage 和 TestTaskPluginAppliance 的关联关系。
这相当于多对多关系的“中间表”。
数据库里现在三张表各有一条相关记录。
1）准备数据（setUp）
插入一条包记录、一条设备记录、一条中间表记录。
2）验证初始状态
确认数据已按预期插入。
3）执行删除方法
调用 delete_with_related()。
4）验证最终状态
所有关联数据都被删光。
5）测试结束回滚
Django 自动清空测试数据库，准备下一个测试
template
js<i-button class="btn" v-if="me.is_superuser" @click="clearTaskRelated()" :type="'error'" :disabled="!me.is_superuser">脏数据清除</i-button>
script
js// 脏数据清除（物模型插件任务/关联表/插件设备表）
async clearTaskRelated() {
const vm = this;
// 显示确认弹窗
await vm.$confirm('您确定要删除该任务及其关联的数据吗？此操作不可撤销!!!')
try {
// 调用后端删除接口
const deleteResp = await vm.api('test_task_package').showLoading(false).post({
action: 'delete_with_related'
}, {
test_task_id: vm.$route.query.task
});
// 检查删除操作的响应
if (deleteResp.status === 204) {
vm.$Message.success({
content: '任务及其关联数据已成功删除。',
duration: 3
});
} else {
vm.$Message.error({
content: deleteResp.data.message,
duration: 5
});
}
} catch (error) {
console.error('Error:', deleteResp.data.message);
vm.$Message.error({
content: '操作失败，请检查网络连接或稍后重试。',
duration: 5
});
}
}
第一个参数 { action: 'delete_with_related' } → 这是后端的 DRF Action 调用方式。
第二个参数 { test_task_id: vm.$route.query.task } → 把任务 ID 传给后端。
showLoading(false)：
禁用全局 loading 动画。
设计要点：
test_task_id 从 URL query 参数中获取（this.$route.query.task），无需额外输入。
API 请求是 POST，符合 DRF @action(methods=['POST']) 要求。
1）操作前确认：用弹窗防止误删，明确提示危险操作。
2）API 调用模式：
链式调用 (vm.api(...).showLoading(false).post(...))。
动作参数 { action: 'delete_with_related' } 对应 DRF 自定义 Action。
3）参数传递：test_task_id 来自 URL query，确保接口参数和后端匹配。
4）状态码处理：
204 表示删除成功。
其他状态用后端返回的 message 提示用户。
5、异常处理：
捕获请求错误，防止 UI 卡死。
提示用户并记录日志。
本文作者:lixf6
本文链接:
版权声明:本博客所有文章除特别声明外，均采用 BY-NC-SA 许可协议。转载请注明出处！