asgiref icon indicating copy to clipboard operation
asgiref copied to clipboard

added AsyncSingleThreadContext

Open Arfey opened this issue 7 months ago • 0 comments

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)()       

Arfey avatar Apr 26 '25 00:04 Arfey