asgiref
asgiref copied to clipboard
added AsyncSingleThreadContext
I have a problem with async database connections for Django TestCase.
The psycopg3 connection is not thread-safe. As a result, I need to create a new connection per thread. This is a problem for the TestCase coz it uses a transaction for each test case.
class TestCase(TransactionTestCase):
async def _apost_teardown(self):
for conn in async_connections.all(initialized_only=True):
await conn.close()
def _setup_and_call(self, result, debug=False):
if iscoroutinefunction(testMethod):
############### convert test
setattr(self, self._testMethodName, async_to_sync(testMethod))
if debug:
super().debug()
else:
super().__call__(result)
self._post_teardown()
############### convert teardown
async_to_sync(self._apost_teardown)()
I can do something like that coz each async_to_sync function uses a new thread executor (and different connection as a result). To solve that, I try to implement something similar for ThreadSensitiveContext.
class TestCase(TransactionTestCase):
def _setup_and_call(self, result, debug=False):
with AsyncSingleThreadContext()
if iscoroutinefunction(testMethod):
############### convert test
setattr(self, self._testMethodName, async_to_sync(testMethod))
if debug:
super().debug()
else:
super().__call__(result)
self._post_teardown()
############### convert teardown
async_to_sync(self._apost_teardown)()