如何在django测试框架中模拟两个并行请求答案

【问题标题】：How to simulate two parallel requests in django testing framework如何在django测试框架中模拟两个并行请求
【发布时间】：2021-09-01 16:58:47
【问题描述】：

我有一个 Django 应用程序（使用 uWSGI 和 nginx，以及原子视图），它的视图可以在数据库（postgres）中创建模型的新项目。在创建任何内容之前，视图会检查数据库中是否不存在该记录，例如：

...
try:
    newfile = DataFile.objects.get(md5=request.POST['md5'])
except DataFile.DoesNotExist:
    newfile = DataFile.objects.create(md5=request.POST['md5'], filename=request.POST['filename'])

return JsonResponse({'file_id': newfile.pk})

我注意到有时这不起作用，并且我在数据库中得到了重复项（这很容易通过唯一约束解决）。如果存在缓存或竞争条件，我不确定为什么会发生这种情况，但我想至少通过 Django 测试框架中的测试来涵盖该行为。但是，我不知道如何模拟两个并行请求。有没有办法在不等待第一个请求的情况下触发下一个请求，内置到框架中，还是应该为此使用多处理或类似方法？

【问题讨论】：

你有没有考虑过使用像get_or_create()这样的原子操作？
@KlausD。我想这段代码有点旧，但我认为如果我正确理解文档，使用它（结合唯一约束）可以解决问题。因为仅使用约束和 try/except 块将需要尝试/排除 IntegrityError 的 .create 行并添加另一个 .get 调用？

标签： python django

【解决方案1】：

我建议你使用异步循环来触发 2 个相当同时的请求。

例子：

async def test_case(request):
    try:
        newfile = DataFile.objects.get(md5=request.POST['md5'])
    except DataFile.DoesNotExist:
        newfile = DataFile.objects.create(md5=request.POST['md5'], filename=request.POST['filename'])

    return JsonResponse({'file_id': newfile.pk})


async def simult(request):
    t_case_0 = await test_case(request)
    t_case_1 = await test_case(request)

asyncio.run(simult(request))

【讨论】：