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 3 commits
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
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
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