Как использовать несколько подключений к базе данных в Django TestCase

Я пишу тестовый пример для воспроизведения тупика в моем сервисе. После того как я использовал multiprocessing для создания двух потоков, я обнаружил, что они используют одно и то же соединение с базой данных. Поэтому я не могу воспроизвести сценарий тупика в тесткейсе.

Как решить эту проблему?

Мой код выглядит следующим образом:

   @transaction.atomic()
    def process_1():
        change = Change.objects.select_for_update().get(id="1")
        time.sleep(5)
        change = Change.objects.select_for_update().get(id="2")

    @transaction.atomic()
    def process_2():
        change = Change.objects.select_for_update().get(id="2")
        time.sleep(5)
        change = Change.objects.select_for_update().get(id="1")

    p1 = Process(target=process_1)
    p2 = Process(target=process_2)
    p1.start()
    p2.start()
    p1.join()
    p2.join()
    self.assertEqual(p1.exitcode, 0)
    self.assertEqual(p2.exitcode, 0)

Это происходит из-за того, что подключения к базе данных Django автоматически не распределяются между процессами при использовании многопроцессорной обработки (каждый процесс должен создавать свое собственное подключение к базе данных).

Просто закройте подключение к базе данных, прежде чем запускать новые процессы:

from django.db import connection

# Here your functions ...

connection.close()  # Ensure each process gets a fresh DB connection

p1 = Process(target=process_1)
p2 = Process(target=process_2)
p1.start()
p2.start()
p1.join()
p2.join()
Вернуться на верх