为什么 Django 模型 return 线程池中的空查询集?
Why did the Django model return an empty queryset in the thread pool?
我在 API 中使用了 ThreadPoolExecutor。在测试用例中,我生成了一个 Movie 实例并将其保存在数据库 (PostgreSQL) 中。我在测试用例中打印了电影计数,API,以及线程池函数。为什么在线程池中 return 一个空查询集?
view.py:
def movie_count():
print(Movie.objects.count(), "movie count in the thread")
class MovieListView(APIView):
renderer_classes = (
render.CamelCaseJSONRenderer,
renderers.BrowsableAPIRenderer,
)
def get(self, request, key, format=None):
print(Movie.objects.count(), "movie count before the thread")
with ThreadPoolExecutor(1) as executor:
bookmarked_future = executor.submit(movie_count)
bookmarked_future.result()
return Response({})
test.py
def test_mock_function(self):
Movie.objects.create(
title="Title"
)
print(Movie.objects.count(), "movie count in the test")
url = reverse('video_api:list', args=('tops',))
self.client.get(url, format='json', data={'limit': 30})
结果:
1 movie counts in the test
1 movie count before the thread
0 movie count in the thread
我找到了答案。
the APITestCase wraps the tests with 2 atomic() blocks, one for the
whole test class and one for each test within the class. This
essentially stops tests from altering the database for other tests as
the transactions are rolled back at the end of each test. By having
this second atomic() block around the whole test class, specific
database transaction behaviour can be hard to test and hence you'd
want to drop back to using APITransactionTestCase.
我在 API 中使用了 ThreadPoolExecutor。在测试用例中,我生成了一个 Movie 实例并将其保存在数据库 (PostgreSQL) 中。我在测试用例中打印了电影计数,API,以及线程池函数。为什么在线程池中 return 一个空查询集?
view.py:
def movie_count():
print(Movie.objects.count(), "movie count in the thread")
class MovieListView(APIView):
renderer_classes = (
render.CamelCaseJSONRenderer,
renderers.BrowsableAPIRenderer,
)
def get(self, request, key, format=None):
print(Movie.objects.count(), "movie count before the thread")
with ThreadPoolExecutor(1) as executor:
bookmarked_future = executor.submit(movie_count)
bookmarked_future.result()
return Response({})
test.py
def test_mock_function(self):
Movie.objects.create(
title="Title"
)
print(Movie.objects.count(), "movie count in the test")
url = reverse('video_api:list', args=('tops',))
self.client.get(url, format='json', data={'limit': 30})
结果:
1 movie counts in the test
1 movie count before the thread
0 movie count in the thread
我找到了答案。
the APITestCase wraps the tests with 2 atomic() blocks, one for the whole test class and one for each test within the class. This essentially stops tests from altering the database for other tests as the transactions are rolled back at the end of each test. By having this second atomic() block around the whole test class, specific database transaction behaviour can be hard to test and hence you'd want to drop back to using APITransactionTestCase.