From b049b9db973c509fc8ff9772899dcb3442c2ab24 Mon Sep 17 00:00:00 2001
From: Roland Sommer <ros@solute.de>
Date: Fri, 1 Apr 2022 12:42:30 +0200
Subject: [PATCH] Fix postgres KeyError with missing explicit columns (#190)

If no explicit colum info is present, the postgres Record class causes a
KeyError. Besides the given example in encode/databases#190 this also
happens if you construct the queries from text-clauses.

Handling this the same way as if there is no entry in `self._column_map`
fixes the error.

A simple testcase is added as well.
---
 databases/backends/postgres.py |  5 ++++-
 tests/test_databases.py        | 18 ++++++++++++++++++
 2 files changed, 22 insertions(+), 1 deletion(-)

diff --git a/databases/backends/postgres.py b/databases/backends/postgres.py
index 3e1a6fff..3e8c9f8b 100644
--- a/databases/backends/postgres.py
+++ b/databases/backends/postgres.py
@@ -140,7 +140,10 @@ def __getitem__(self, key: typing.Any) -> typing.Any:
         elif isinstance(key, int):
             idx, datatype = self._column_map_int[key]
         else:
-            idx, datatype = self._column_map[key]
+            try:
+                idx, datatype = self._column_map[key]
+            except KeyError:
+                return self._row[key]
         raw = self._row[idx]
         processor = datatype._cached_result_processor(self._dialect, None)
 
diff --git a/tests/test_databases.py b/tests/test_databases.py
index 7a0b84fd..7b599293 100644
--- a/tests/test_databases.py
+++ b/tests/test_databases.py
@@ -222,6 +222,24 @@ async def test_queries(database_url):
             assert iterate_results[2]["completed"] == True
 
 
+@pytest.mark.parametrize("database_url", DATABASE_URLS)
+@mysql_versions
+@async_adapter
+async def test_queries_manual(database_url):
+    async with Database(database_url) as database:
+        async with database.transaction(force_rollback=True):
+            _t = sqlalchemy.sql.text
+
+            query = notes.insert()
+            values = {"text": "example1", "completed": True}
+            await database.execute(query, values)
+
+            query = sqlalchemy.select([_t("n.text")], from_obj=[_t("notes n")])
+            result = await database.fetch_one(query)
+
+            assert dict(result)
+
+
 @pytest.mark.parametrize("database_url", DATABASE_URLS)
 @mysql_versions
 @async_adapter