Skip to content
This repository was archived by the owner on Aug 19, 2025. It is now read-only.
Open
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Next Next commit
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 #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.
  • Loading branch information
Roland Sommer committed Apr 1, 2022
commit b049b9db973c509fc8ff9772899dcb3442c2ab24
5 changes: 4 additions & 1 deletion databases/backends/postgres.py
Original file line number Diff line number Diff line change
Expand Up @@ -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]
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we be setting raw to self._row[key] to ensure the processor runs on it still?

raw = self._row[idx]
processor = datatype._cached_result_processor(self._dialect, None)

Expand Down
18 changes: 18 additions & 0 deletions tests/test_databases.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down