Handling raw response from sql drivers (as part of #9)

Author: gvbgduh

databases/backends/postgres.py

@@ -74,18 +74,21 @@ def __init__(self, row: tuple, result_columns: tuple, dialect: Dialect) -> None:
             for idx, (column_name, _, _, datatype) in enumerate(self._result_columns)
         }
 
-    def __getitem__(self, key: str) -> typing.Any:
-        idx, datatype = self._column_map[key]
-        raw = self._row[idx]
+    def __getitem__(self, key: typing.Any) -> typing.Any:
         try:
-            processor = _result_processors[datatype]
+            idx, datatype = self._column_map[key]
+            raw = self._row[idx]
+            try:
+                processor = _result_processors[datatype]
+            except KeyError:
+                processor = datatype.result_processor(self._dialect, None)
+                _result_processors[datatype] = processor
+
+            if processor is not None:
+                return processor(raw)
+            return raw
         except KeyError:
-            processor = datatype.result_processor(self._dialect, None)
-            _result_processors[datatype] = processor
-
-        if processor is not None:
-            return processor(raw)
-        return raw
+            return self._row[key]
 
     def __iter__(self) -> typing.Iterator:
         return iter(self._column_map)

tests/test_databases.py

@@ -587,5 +587,37 @@ async def test_queries_with_expose_backend_connection(database_url):
             assert result[1] == "example1"
             assert result[2] == True
 
-# TODO unittests at the connection level
-# TODO (?) Double-check connections are released back to the pool
+
+@pytest.mark.parametrize("database_url", DATABASE_URLS)
+@async_adapter
+async def test_queries_with_sqlalchemy_test(database_url):
+    """
+    Test for inserting and retriving the data using the `sqlalchemy.text` query object.
+    """
+    async with Database(database_url) as database:
+        async with database.transaction(force_rollback=True):
+            # Insert query
+            insert_query = "INSERT INTO notes (text, completed) VALUES (:text, :completed)"
+            
+            # execute_many()
+            query = sqlalchemy.text(insert_query)
+            values = [("example1", True), ("example2", False), ("example3", True)]
+            for text, completed in values:
+                current_query = query.bindparams(text=text, completed=completed)
+                await database.execute(current_query)
+
+            # Select query
+            select_query = "SELECT notes.id, notes.text, notes.completed FROM notes"
+
+            # fetch_all()
+            query = sqlalchemy.text(select_query)
+            results = await database.fetch_all(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