Remove non-working pooling, and any reference to single_connection.

Author: grigi

tortoise/backends/asyncpg/client.py

@@ -8,9 +8,9 @@
 from tortoise.backends.asyncpg.executor import AsyncpgExecutor
 from tortoise.backends.asyncpg.schema_generator import AsyncpgSchemaGenerator
 from tortoise.backends.base.client import (BaseDBAsyncClient, BaseTransactionWrapper,
-                                           ConnectionWrapper, SingleConnectionWrapper)
-from tortoise.exceptions import (ConfigurationError, DBConnectionError, IntegrityError,
-                                 OperationalError, TransactionManagementError)
+                                           ConnectionWrapper)
+from tortoise.exceptions import (DBConnectionError, IntegrityError, OperationalError,
+                                 TransactionManagementError)
 from tortoise.transactions import current_transaction_map
 
 
@@ -49,7 +49,7 @@ def __init__(self, user: str, password: str, database: str, host: str, port: Sup
             port=self.port,
             database=self.database
         )
-        self._db_pool = None  # Type: Optional[asyncpg.pool.Pool]
+        # self._db_pool = None  # Type: Optional[asyncpg.pool.Pool]
         self._connection = None  # Type: Optional[asyncpg.Connection]
 
         self._transaction_class = type(
@@ -58,10 +58,7 @@ def __init__(self, user: str, password: str, database: str, host: str, port: Sup
 
     async def create_connection(self) -> None:
         try:
-            if not self.single_connection:
-                self._db_pool = await asyncpg.create_pool(self.dsn)
-            else:
-                self._connection = await asyncpg.connect(self.dsn)
+            self._connection = await asyncpg.connect(self.dsn)
             self.log.debug(
                 'Created connection with params: user=%s database=%s host=%s port=%s',
                 self.user, self.database, self.host, self.port
@@ -72,14 +69,11 @@ async def create_connection(self) -> None:
             ))
 
     async def close(self) -> None:
-        if self._db_pool:
-            await self._db_pool.close()
         if self._connection:
             await self._connection.close()
+            self._connection = None
 
     async def db_create(self) -> None:
-        single_connection = self.single_connection
-        self.single_connection = True
         self._connection = await asyncpg.connect(self.DSN_TEMPLATE.format(
             user=self.user,
             password=self.password,
@@ -91,11 +85,8 @@ async def db_create(self) -> None:
             'CREATE DATABASE "{}" OWNER "{}"'.format(self.database, self.user)
         )
         await self._connection.close()  # type: ignore
-        self.single_connection = single_connection
 
     async def db_delete(self) -> None:
-        single_connection = self.single_connection
-        self.single_connection = True
         self._connection = await asyncpg.connect(self.DSN_TEMPLATE.format(
             user=self.user,
             password=self.password,
@@ -105,22 +96,15 @@ async def db_delete(self) -> None:
         ))
         try:
             await self.execute_script('DROP DATABASE "{}"'.format(self.database))
-        except asyncpg.InvalidCatalogNameError:
+        except asyncpg.InvalidCatalogNameError:  # pragma: nocoverage
             pass
         await self._connection.close()  # type: ignore
-        self.single_connection = single_connection
 
     def acquire_connection(self) -> ConnectionWrapper:
-        if not self.single_connection:
-            return self._db_pool.acquire()  # type: ignore
-        else:
-            return ConnectionWrapper(self._connection)
+        return ConnectionWrapper(self._connection)
 
     def _in_transaction(self) -> 'TransactionWrapper':
-        if self.single_connection:
-            return self._transaction_class(self.connection_name, connection=self._connection)
-        else:
-            return self._transaction_class(self.connection_name, pool=self._db_pool)
+        return self._transaction_class(self.connection_name, self._connection)
 
     @translate_exceptions
     async def execute_insert(self, query: str, values: list) -> int:
@@ -142,29 +126,11 @@ async def execute_script(self, query: str) -> None:
         async with self.acquire_connection() as connection:
             await connection.execute(query)
 
-    async def get_single_connection(self):
-        if self.single_connection:
-            return self._single_connection_class(self.connection_name, self._connection)
-        else:
-            connection = await self._db_pool._acquire(None)
-            return self._single_connection_class(self.connection_name, connection)
-
-    async def release_single_connection(self, single_connection):
-        if not self.single_connection:
-            await self._db_pool.release(single_connection.connection)
-
 
 class TransactionWrapper(AsyncpgDBClient, BaseTransactionWrapper):
-    def __init__(self, connection_name: str, pool=None, connection=None) -> None:
-        if pool and connection:
-            raise ConfigurationError('You must pass either connection or pool')
+    def __init__(self, connection_name: str, connection) -> None:
         self._connection = connection
         self.log = logging.getLogger('db_client')
-        self._pool = pool
-        self.single_connection = True
-        self._single_connection_class = type(
-            'SingleConnectionWrapper', (SingleConnectionWrapper, self.__class__), {}
-        )
         self._transaction_class = self.__class__
         self._old_context_value = None
         self.connection_name = connection_name
@@ -173,12 +139,7 @@ def __init__(self, connection_name: str, pool=None, connection=None) -> None:
     def acquire_connection(self) -> ConnectionWrapper:
         return ConnectionWrapper(self._connection)
 
-    async def _get_connection(self):
-        return await self._pool._acquire(None)
-
     async def start(self):
-        if not self._connection:
-            self._connection = await self._get_connection()
         self.transaction = self._connection.transaction()
         await self.transaction.start()
         current_transaction = current_transaction_map[self.connection_name]
@@ -190,17 +151,11 @@ async def commit(self):
             await self.transaction.commit()
         except asyncpg.exceptions._base.InterfaceError as exc:
             raise TransactionManagementError(exc)
-        if self._pool:
-            await self._pool.release(self._connection)
-            self._connection = None
         current_transaction_map[self.connection_name].set(self._old_context_value)
 
     async def rollback(self):
         try:
             await self.transaction.rollback()
         except asyncpg.exceptions._base.InterfaceError as exc:
             raise TransactionManagementError(exc)
-        if self._pool:
-            await self._pool.release(self._connection)
-            self._connection = None
         current_transaction_map[self.connection_name].set(self._old_context_value)

tortoise/backends/asyncpg/executor.py

@@ -8,6 +8,6 @@
 class AsyncpgExecutor(BaseExecutor):
     def _prepare_insert_statement(self, columns: List[str]) -> str:
         return str(
-            self.connection.query_class.into(Table(self.model._meta.table)).columns(*columns)
+            self.db.query_class.into(Table(self.model._meta.table)).columns(*columns)
             .insert('???').returning('id')
         ).replace("'???'", ','.join(['$%d' % (i + 1, ) for i in range(len(columns))]))

tortoise/backends/base/client.py

@@ -12,13 +12,9 @@ class BaseDBAsyncClient:
     executor_class = BaseExecutor
     schema_generator = BaseSchemaGenerator
 
-    def __init__(self, connection_name: str, single_connection: bool = True, **kwargs) -> None:
+    def __init__(self, connection_name: str, **kwargs) -> None:
         self.log = logging.getLogger('db_client')
-        self.single_connection = single_connection
         self.connection_name = connection_name
-        self._single_connection_class = type(
-            'SingleConnectionWrapper', (SingleConnectionWrapper, self.__class__), {}
-        )
 
     async def create_connection(self) -> None:
         raise NotImplementedError()  # pragma: nocoverage
@@ -47,12 +43,6 @@ async def execute_query(self, query: str) -> Sequence[dict]:
     async def execute_script(self, query: str) -> None:
         raise NotImplementedError()  # pragma: nocoverage
 
-    async def get_single_connection(self) -> 'BaseDBAsyncClient':
-        raise NotImplementedError()  # pragma: nocoverage
-
-    async def release_single_connection(self, single_connection: 'BaseDBAsyncClient') -> None:
-        raise NotImplementedError()  # pragma: nocoverage
-
 
 class ConnectionWrapper:
     __slots__ = ('connection', )
@@ -67,35 +57,6 @@ async def __aexit__(self, exc_type, exc_val, exc_tb):
         pass
 
 
-class SingleConnectionWrapper(BaseDBAsyncClient):
-    # pylint: disable=W0223,W0231
-
-    def __init__(self, connection_name: str, connection, closing_callback=None) -> None:
-        self.log = logging.getLogger('db_client')
-        self.connection_name = connection_name
-        self.connection = connection
-        self.single_connection = True
-        self.closing_callback = closing_callback
-
-    def acquire_connection(self) -> ConnectionWrapper:
-        return ConnectionWrapper(self.connection)
-
-    async def get_single_connection(self) -> 'SingleConnectionWrapper':
-        # Real class object is generated in runtime, so we use __class__ reference
-        # instead of using SingleConnectionWrapper directly
-        return self.__class__(self.connection_name, self.connection, self)
-
-    async def release_single_connection(self, single_connection: 'BaseDBAsyncClient') -> None:
-        return
-
-    async def __aenter__(self):
-        return self
-
-    async def __aexit__(self, exc_type, exc_val, exc_tb):
-        if self.closing_callback:
-            await self.closing_callback(self)
-
-
 class BaseTransactionWrapper:
     async def start(self) -> None:
         raise NotImplementedError()  # pragma: nocoverage

tortoise/backends/base/config_generator.py

@@ -57,7 +57,6 @@ def expand_db_url(db_url: str, testing: bool = False) -> dict:
         params[key] = val[-1]
 
     if testing:
-        params['single_connection'] = True
         path = path.replace('\\{', '{').replace('\\}', '}')
         path = path.format(uuid.uuid4().hex)
 

tortoise/backends/base/executor.py

@@ -15,13 +15,11 @@ class BaseExecutor:
     def __init__(self, model, db=None, prefetch_map=None, prefetch_queries=None):
         self.model = model
         self.db = db
-        self.connection = None
         self.prefetch_map = prefetch_map if prefetch_map else {}
         self._prefetch_queries = prefetch_queries if prefetch_queries else {}
 
     async def execute_select(self, query, custom_fields=None):
-        self.connection = await self.db.get_single_connection()
-        raw_results = await self.connection.execute_query(str(query))
+        raw_results = await self.db.execute_query(str(query))
         instance_list = []
         for row in raw_results:
             instance = self.model(**row)
@@ -30,8 +28,6 @@ async def execute_select(self, query, custom_fields=None):
                     setattr(instance, field, row[field])
             instance_list.append(instance)
         await self._execute_prefetch_queries(instance_list)
-        await self.db.release_single_connection(self.connection)
-        self.connection = None
         return instance_list
 
     def _prepare_insert_columns(self):
@@ -60,7 +56,6 @@ def _prepare_insert_statement(self, columns: List[str]) -> str:
         raise NotImplementedError()  # pragma: nocoverage
 
     async def execute_insert(self, instance):
-        self.connection = await self.db.get_single_connection()
         key = '{}:{}'.format(self.db.connection_name, self.model._meta.table)
         if key not in INSERT_CACHE:
             regular_columns, columns = self._prepare_insert_columns()
@@ -73,15 +68,12 @@ async def execute_insert(self, instance):
             instance=instance,
             regular_columns=regular_columns,
         )
-        instance.id = await self.connection.execute_insert(query, values)
-        await self.db.release_single_connection(self.connection)
-        self.connection = None
+        instance.id = await self.db.execute_insert(query, values)
         return instance
 
     async def execute_update(self, instance):
-        self.connection = await self.db.get_single_connection()
         table = Table(self.model._meta.table)
-        query = self.connection.query_class.update(table)
+        query = self.db.query_class.update(table)
         for field, db_field in self.model._meta.fields_db_projection.items():
             field_object = self.model._meta.fields_map[field]
             if not field_object.generated:
@@ -90,9 +82,7 @@ async def execute_update(self, instance):
                     self._field_to_db(field_object, getattr(instance, field), instance)
                 )
         query = query.where(table.id == instance.id)
-        await self.connection.execute_query(str(query))
-        await self.db.release_single_connection(self.connection)
-        self.connection = None
+        await self.db.execute_query(str(query))
         return instance
 
     async def execute_delete(self, instance):
@@ -133,7 +123,7 @@ async def _prefetch_m2m_relation(self, instance_list, field, related_query):
 
         through_table = Table(field_object.through)
 
-        subquery = self.connection.query_class.from_(through_table).select(
+        subquery = self.db.query_class.from_(through_table).select(
             getattr(through_table, field_object.backward_key).as_('_backward_relation_key'),
             getattr(through_table, field_object.forward_key).as_('_forward_relation_key')
         ).where(getattr(through_table, field_object.backward_key).isin(instance_id_set))
@@ -145,12 +135,12 @@ async def _prefetch_m2m_relation(self, instance_list, field, related_query):
             subquery._backward_relation_key.as_('_backward_relation_key'),
             *[getattr(related_query_table, field).as_(field) for field in related_query.fields]
         )
-        raw_results = await self.connection.execute_query(str(query))
+        raw_results = await self.db.execute_query(str(query))
         relations = {(e['_backward_relation_key'], e['id']) for e in raw_results}
         related_object_list = [related_query.model(**e) for e in raw_results]
         await self.__class__(
             model=related_query.model,
-            db=self.connection,
+            db=self.db,
             prefetch_map=related_query._prefetch_map,
         ).fetch_for_list(related_object_list)
         related_object_map = {e.id: e for e in related_object_list}
@@ -188,7 +178,7 @@ def _make_prefetch_queries(self):
             else:
                 related_model_field = self.model._meta.fields_map.get(field)
                 related_model = related_model_field.type
-                related_query = related_model.all().using_db(self.connection)
+                related_query = related_model.all().using_db(self.db)
             if forwarded_prefetches:
                 related_query = related_query.prefetch_related(*forwarded_prefetches)
             self._prefetch_queries[field] = related_query
@@ -209,7 +199,6 @@ async def _execute_prefetch_queries(self, instance_list):
         return instance_list
 
     async def fetch_for_list(self, instance_list, *args):
-        self.connection = await self.db.get_single_connection()
         self.prefetch_map = {}
         for relation in args:
             relation_split = relation.split('__')
@@ -223,8 +212,6 @@ async def fetch_for_list(self, instance_list, *args):
             if forwarded_prefetch:
                 self.prefetch_map[first_level_field].add(forwarded_prefetch)
         await self._execute_prefetch_queries(instance_list)
-        await self.db.release_single_connection(self.connection)
-        self.connection = None
         return instance_list
 
     @classmethod