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

Question

我有一个 Django 应用程序(使用 uWSGI 和 nginx，以及原 subview )，其 View 在数据库 (postgres) 中创建模型的新项目。在创建任何内容之前， View 会检查数据库中是否不存在该记录，例如:

...
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 测试框架中的测试来涵盖这种行为。但是，我不知道如何模拟两个并行请求。有没有一种方法可以在不等待框架内置的第一个请求的情况下触发下一个请求，或者应该为此使用多处理或类似方法？

Answer 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))