16

私は、PostgreSQL データベースにクエリを実行する必要がある Django Web アプリケーションで作業しています。Pythonスレッドインターフェイスを使用して同時実行を実装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例外が返されています。

この単体テストの実行結果は次のようになります。

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

これはすべて、SQLite などではなく、スレッドセーフであるはずの PostgreSQL データベースで発生していることを思い出してください。PostgreSQL も使用してテストを実行しました。

この時点で、何が失敗する可能性があるかについて完全に迷っています。アイデアや提案はありますか？

ありがとう！

編集：テストでうまくいくかどうかを確認するためだけに、小さなビューを書きました。ビューのコードは次のとおりです。

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 を呼び出した後、最終的にビューに都市が表示されます。

したがって、スレッドを使用してクエリを作成することは、コードをテストする必要がある場合にのみ問題になると思います。本番環境では、問題なく動作します。

この種のコードを正常にテストするための有用な提案はありますか?

4

3 に答える 3

16

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

class ThreadingTest(TransactionTestCase):

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

TransactionTestCase と TestCase は、データベースを既知の状態にリセットする方法と、テスト コードでコミットとロールバックの効果をテストする機能以外は同じです。TransactionTestCase は、すべてのテーブルを切り捨てて初期データをリロードすることにより、テストを実行する前にデータベースをリセットします。TransactionTestCase は commit と rollback を呼び出し、これらの呼び出しがデータベースに及ぼす影響を観察します。

于 2012-06-08T13:16:35.033 に答える
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.
    """

現在、トランザクションは TestCase 内でコミットされていないため、変更は他のスレッドには表示されません。

于 2014-02-22T07:13:43.230 に答える
2

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

于 2012-06-08T12:09:30.473 に答える