Как имитировать два параллельных запроса во фреймворке для тестирования django

У меня есть приложение 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. Однако я не знаю, как имитировать два параллельных запроса. Есть ли способ запускать следующий запрос, не дожидаясь первого, встроенный во фреймворк, или для этого нужно использовать мультипроцессинг или что-то подобное?

Я предлагаю вам использовать цикл async для запуска 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))
Вернуться на верх