Get rid of experimental solution and minor fixes.

Author: gvbgduh

databases/backends/mysql.py

@@ -10,7 +10,7 @@
 from sqlalchemy.sql import ClauseElement
 from sqlalchemy.types import TypeEngine
 
-from databases.core import DatabaseURL, NoBackendMethod
+from databases.core import DatabaseURL
 from databases.interfaces import ConnectionBackend, DatabaseBackend, TransactionBackend
 
 logger = logging.getLogger("databases")
@@ -135,25 +135,6 @@ async def execute_many(self, query: ClauseElement, values: list) -> None:
         finally:
             await cursor.close()
 
-    async def raw_api_call(self, method: str, *args: typing.Any, **kwargs: typing.Any) -> typing.Any:
-        """
-        NOTE: highly experimental, seems to be a dead-end for generalized solution 
-        """
-        assert self._connection is not None, "Connection is not acquired"
-        cursor = await self._connection.cursor()
-        try:
-            if 'execute' not in method:
-                await cursor.execute(*args, **kwargs)
-                api_method = getattr(cursor, method)
-                return await api_method()
-            else:
-                api_method = getattr(cursor, method)
-                return await api_method(*args, **kwargs)
-        except AttributeError:
-            raise NoBackendMethod(f'{self.database._backend._dialect.driver} has no "{method}" implemented.')
-        finally:
-            await cursor.close()
-
     async def expose_backend_connection(self) -> aiomysql.connection.Connection:
         assert self._connection is not None, "Connection is not acquired"
         return self._connection

databases/backends/postgres.py

@@ -7,7 +7,7 @@
 from sqlalchemy.engine.interfaces import Dialect
 from sqlalchemy.sql import ClauseElement
 
-from databases.core import DatabaseURL, NoBackendMethod
+from databases.core import DatabaseURL
 from databases.interfaces import ConnectionBackend, DatabaseBackend, TransactionBackend
 
 logger = logging.getLogger("databases")
@@ -144,17 +144,6 @@ async def execute_many(self, query: ClauseElement, values: list) -> None:
             single_query, args, result_columns = self._compile(single_query)
             await self._connection.execute(single_query, *args)
 
-    async def raw_api_call(self, method: str, *args: typing.Any, **kwargs: typing.Any) -> typing.Any:
-        """
-        NOTE: highly experimental, seems to be a dead-end for generalized solution 
-        """
-        assert self._connection is not None, "Connection is not acquired"
-        try:
-            api_method = getattr(self._connection, method)
-            return await api_method(*args, **kwargs)
-        except AttributeError:
-            raise NoBackendMethod(f'{self.database._backend._dialect.driver} has no "{method}" implemented.')
-
     async def expose_backend_connection(self) -> asyncpg.connection.Connection:
         assert self._connection is not None, "Connection is not acquired"
         return self._connection

databases/backends/sqlite.py

@@ -9,7 +9,7 @@
 from sqlalchemy.sql import ClauseElement
 from sqlalchemy.types import TypeEngine
 
-from databases.core import DatabaseURL, NoBackendMethod
+from databases.core import DatabaseURL
 from databases.interfaces import ConnectionBackend, DatabaseBackend, TransactionBackend
 
 logger = logging.getLogger("databases")
@@ -116,17 +116,6 @@ async def execute_many(self, query: ClauseElement, values: list) -> None:
         for value in values:
             await self.execute(query, value)
 
-    async def raw_api_call(self, method: str, *args: typing.Any, **kwargs: typing.Any) -> typing.Any:
-        """
-        NOTE: highly experimental, seems to be a dead-end for generalized solution 
-        """
-        assert self._connection is not None, "Connection is not acquired"
-        try:
-            api_method = getattr(self._connection, method)
-            return await api_method(*args, **kwargs)
-        except AttributeError:
-            raise NoBackendMethod(f'{self.database._backend._dialect.driver} has no "{method}" implemented.')
-
     async def expose_backend_connection(self) -> aiosqlite.core.Connection:
         assert self._connection is not None, "Connection is not acquired"
         return self._connection

databases/core.py

@@ -104,10 +104,6 @@ async def execute_many(self, query: ClauseElement, values: list) -> None:
         async with self.connection() as connection:
             return await connection.execute_many(query=query, values=values)
 
-    async def raw_api_call(self, method: str, *args: typing.Any, **kwargs: typing.Any) -> typing.Any:
-        async with self.connection() as connection:
-            return await connection.raw_api_call(method, *args, **kwargs)
-
     async def expose_backend_connection(self) -> typing.Any:
         async with self.connection() as connection:
             return await connection.expose_backend_connection()
@@ -176,9 +172,6 @@ async def execute(self, query: ClauseElement, values: dict = None) -> typing.Any
     async def execute_many(self, query: ClauseElement, values: list) -> None:
         await self._connection.execute_many(query, values)
 
-    async def raw_api_call(self, method: str, *args: typing.Any, **kwargs: typing.Any) -> typing.Any:
-        return await self._connection.raw_api_call(method, *args, **kwargs)
-
     async def expose_backend_connection(self) -> typing.Any:
         return await self._connection.expose_backend_connection()
 
@@ -362,7 +355,3 @@ def __repr__(self) -> str:
         if self.password:
             url = str(self.replace(password="********"))
         return f"{self.__class__.__name__}({repr(url)})"
-
-
-class NoBackendMethod(Exception):
-    pass

databases/interfaces.py

@@ -33,6 +33,9 @@ async def execute(self, query: ClauseElement, values: dict = None) -> typing.Any
     async def execute_many(self, query: ClauseElement, values: list) -> None:
         raise NotImplementedError()  # pragma: no cover
 
+    async def expose_backend_connection(self) -> typing.Any:
+        raise NotImplementedError()  # pragma: no cover
+
     async def iterate(
         self, query: ClauseElement
     ) -> typing.AsyncGenerator[typing.Mapping, None]:

tests/test_databases.py

@@ -516,62 +516,6 @@ async def get_connection_2():
         await task_2
 
 
-@pytest.mark.parametrize("database_url", DATABASE_URLS)
-@async_adapter
-async def test_queries_with_raw_api_call(database_url):
-    """
-    Test that the basic `execute()`, `execute_many()`, `fetch_all()``, and
-    `fetch_one()` interfaces are working as expected being called as raw driver calls.
-    """
-    async with Database(database_url) as database:
-        async with database.transaction(force_rollback=True):
-            # Insert query
-            if str(database_url).startswith('mysql'):
-                insert_query = "INSERT INTO notes (text, completed) VALUES (%s, %s)"
-            else:
-                insert_query = "INSERT INTO notes (text, completed) VALUES ($1, $2)"
-
-            # execute()
-            values = ("example1", True)
-
-            if str(database_url).startswith('postgresql'):
-                await database.raw_api_call('execute', insert_query, *values)
-            else:
-                await database.raw_api_call('execute', insert_query, values)
-
-            # execute_many()
-            values = [("example2", False), ("example3", True)]
-            await database.raw_api_call('executemany', insert_query, values)
-
-            # Select query
-            select_query = "SELECT notes.id, notes.text, notes.completed FROM notes"
-
-            # fetch_all()
-            if str(database_url).startswith('postgresql'):
-                results = await database.raw_api_call('fetch', select_query)
-            elif str(database_url).startswith('mysql'):
-                results = await database.raw_api_call('fetchall', select_query)
-            elif str(database_url).startswith('sqlite'):
-                results = await database.raw_api_call('execute_fetchall', select_query)
-
-            assert len(results) == 3
-            # Raw output for the raw request
-            assert results[0][1] == "example1"
-            assert results[0][2] == True
-            assert results[1][1] == "example2"
-            assert results[1][2] == False
-            assert results[2][1] == "example3"
-            assert results[2][2] == True
-
-            # # fetch_one()
-            # if str(database_url).startswith('postgresql'):
-            #     result = await database.raw_api_call('fetchrow', select_query)
-            # else:
-            #     result = await database.raw_api_call('fetchone', select_query)
-            # assert result[1] == "example1"
-            # assert result[2] == True
-
-
 @pytest.mark.parametrize("database_url", DATABASE_URLS)
 @async_adapter
 async def test_queries_with_expose_backend_connection(database_url):