スレッドの使用時のDjangoのデータベースエラー

Question

私はPostgreSQLデータベースを照会する必要があるDjango Webアプリケーションで作業しています。 Python threadingインターフェイスを使用して並行処理を実装する場合、クエリされた項目にはエラーDoesNotExistが発生しています。もちろん、これらのエラーは、クエリを順次実行するときには発生しません。

私は予期しない動作を実証するために書いたユニットテストを表示してみましょう：

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, []) 

あなたが見ることができるように、最初のテストは確かに問題なく動作しているいくつかのデータベース要求を順次、実行します。しかし、2番目のテストではまったく同じ要求が実行されますが、各要求はスレッドに生成されます。これは実際には失敗し、DoesNotExistの例外を返します。

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')... 

は、このすべてがあることを忘れないでください...

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. 

...他のスレッドが同じような出力を返します。

このユニットテストの実行の出力は次のようですスレッドセーフであるはずのPostgreSQLデータベースで起こります.SQLiteやsimilarsではそうではありません。テストは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を呼び出した後に都市を表示します。

スレッドを使用してクエリを作成することは、コードをテストする必要がある場合にのみ問題になると思います。プロダクションでは問題なく動作します。

この種のコードを正常にテストするための参考にしてください。

Answer 1

これはトランザクションの問題のようです。現在のリクエスト（またはテスト）内に要素を作成している場合、他のスレッドの別の接続からアクセスできないコミットされていないトランザクションがほぼ確実に発生します。これを動作させるには、おそらくmanage your transctions manuallyが必要です。

Answer 2

TransactionTestCaseを使用してみてください：

class ThreadingTest(TransactionTestCase): 

のTestCaseは、メモリ内のデータを保持し、データベースにCOMMITを発行しません。おそらく、スレッドはDBに直接接続しようとしていますが、データはまだコミットされていません。ここでSeedescription： https://docs.djangoproject.com/en/dev/topics/testing/?from=olddocs#django.test.TransactionTestCase

TransactionTestCaseとテストケースでは、データベースは、既知の状態と、コミットとロールバックの効果を試験するため テストコードの能力にリセットされる方法 を除いて同一です。 A TransactionTestCaseは、テストが実行される前にデータベースをリセットし、 すべてのテーブルを切り捨て、初期データを再ロードします。 A TransactionTestCaseは、コミットとロールバックを呼び出し、これらの呼び出しがデータベースに及ぼす影響を確認します。

Answer 3

は、したがって変更が他のスレッドには表示されません、ドキュメントさて、トランザクションはテストケースの内側にコミットされていない

class LiveServerTestCase(TransactionTestCase): 
    """ 
    ... 
    Note that it inherits from TransactionTestCase instead of TestCase because 
    the threads do not share the same transactions (unless if using in-memory 
    sqlite) and each thread needs to commit all their transactions so that the 
    other thread can see the changes. 
    """ 

のこの部分からより明確になります。