Ошибки базы данных в Django при использовании потоков

Я работаю в веб-приложении Django, которое должно запрашивать базу данных PostgreSQL. При реализации concurrency с использованием интерфейса Python threading я получаю ошибки DoesNotExist для запрошенных элементов. Конечно, эти ошибки не возникают при выполнении запросов последовательно.

Позвольте мне показать unit test, который я написал, чтобы продемонстрировать неожиданное поведение:

class ThreadingTest(TestCase):
    fixtures = ['demo_city',]

    def test_sequential_requests(self):
        """
        A very simple request to database, made sequentially.

        A fixture for the cities has been loaded above. It is supposed to be
        six cities in the testing database now. We will made a request for
        each one of the cities sequentially.
        """
        for number in range(1, 7):
            c = City.objects.get(pk=number)
            self.assertEqual(c.pk, number)

    def test_threaded_requests(self):
        """
        Now, to test the threaded behavior, we will spawn a thread for
        retrieving each city from the database.
        """

        threads = []
        cities = []

        def do_requests(number):
            cities.append(City.objects.get(pk=number))

        [threads.append(threading.Thread(target=do_requests, args=(n,))) for n in range(1, 7)]

        [t.start() for t in threads]
        [t.join() for t in threads]

        self.assertNotEqual(cities, [])

Как вы можете видеть, первый тест последовательно выполняет несколько запросов к базе данных, которые действительно работают без проблем. Второй тест, однако, выполняет точно такие же запросы, но каждый запрос порождается в потоке. Это фактически не работает, возвращая исключение DoesNotExist.

Результат выполнения этих модульных тестов выглядит следующим образом:

test_sequential_requests (cesta.core.tests.threadbase.ThreadingTest) ... ok
test_threaded_requests (cesta.core.tests.threadbase.ThreadingTest) ...

Exception in thread Thread-1:
Traceback (most recent call last):
  File "/usr/lib/python2.6/threading.py", line 532, in __bootstrap_inner
    self.run()
  File "/usr/lib/python2.6/threading.py", line 484, in run
    self.__target(*self.__args, **self.__kwargs)
  File "/home/jose/Work/cesta/trunk/src/cesta/core/tests/threadbase.py", line 45, in do_requests
    cities.append(City.objects.get(pk=number))
  File "/home/jose/Work/cesta/trunk/parts/django/django/db/models/manager.py", line 132, in get
    return self.get_query_set().get(*args, **kwargs)
  File "/home/jose/Work/cesta/trunk/parts/django/django/db/models/query.py", line 349, in get
    % self.model._meta.object_name)
DoesNotExist: City matching query does not exist.

... другие потоки возвращают аналогичный результат...

Exception in thread Thread-6:
Traceback (most recent call last):
  File "/usr/lib/python2.6/threading.py", line 532, in __bootstrap_inner
    self.run()
  File "/usr/lib/python2.6/threading.py", line 484, in run
    self.__target(*self.__args, **self.__kwargs)
  File "/home/jose/Work/cesta/trunk/src/cesta/core/tests/threadbase.py", line 45, in do_requests
    cities.append(City.objects.get(pk=number))
  File "/home/jose/Work/cesta/trunk/parts/django/django/db/models/manager.py", line 132, in get
    return self.get_query_set().get(*args, **kwargs)
  File "/home/jose/Work/cesta/trunk/parts/django/django/db/models/query.py", line 349, in get
    % self.model._meta.object_name)
DoesNotExist: City matching query does not exist.


FAIL

======================================================================
FAIL: test_threaded_requests (cesta.core.tests.threadbase.ThreadingTest)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/home/jose/Work/cesta/trunk/src/cesta/core/tests/threadbase.py", line 52, in test_threaded_requests
    self.assertNotEqual(cities, [])
AssertionError: [] == []

----------------------------------------------------------------------
Ran 2 tests in 0.278s

FAILED (failures=1)
Destroying test database for alias 'default' ('test_cesta')...

Помните, что все это происходит в базе данных PostgreSQL, которая должна быть потокобезопасной, а не с SQLite или подобными. Тест также запускался с использованием PostgreSQL.

В этот момент я полностью потерял то, что может быть неудачным. Любая идея или предложение?

Спасибо!

EDIT: Я написал небольшой обзор, чтобы проверить, не работает ли он из тестов. Вот код представления:

def get_cities(request):
    queue = Queue.Queue()

    def get_async_cities(q, n):
        city = City.objects.get(pk=n)
        q.put(city)

    threads = [threading.Thread(target=get_async_cities, args=(queue, number)) for number in range(1, 5)]

    [t.start() for t in threads]
    [t.join() for t in threads]

    cities = list()

    while not queue.empty():
        cities.append(queue.get())

    return render_to_response('async/cities.html', {'cities': cities},
        context_instance=RequestContext(request))

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

Результат заключается в том, что код работает хорошо, запросы успешно выполняются в потоках, и в результате, наконец, отображаются города после вызова его URL-адреса.

Итак, я думаю, что создание запросов с использованием потоков будет проблемой только тогда, когда вам нужно протестировать код. В производстве он будет работать без проблем.

Какие-нибудь полезные предложения для успешного тестирования этого типа кода?