Как использовать несколько подключений к базе данных в 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)

This is happening cause Django’s database connections are not automatically shared across processes when using multiprocessing (each process should create its own database connection).

Just close the database connection before forking the new processes:

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()
Вернуться на верх