Make fetch_val call fetch_one for type conversion (#246)

* Make fetch_val call fetch_one for type conversion

Signed-off-by: Lou Marvin Caraig <[email protected]>

* Add fetch_val tests

* Add the comment to explain the replacement of fetchval

Co-authored-by: Vadim Markovtsev <[email protected]>

Author: se7entyse7en

databases/backends/postgres.py

@@ -178,9 +178,17 @@ async def fetch_one(self, query: ClauseElement) -> typing.Optional[typing.Mappin
     async def fetch_val(
         self, query: ClauseElement, column: typing.Any = 0
     ) -> typing.Any:
-        assert self._connection is not None, "Connection is not acquired"
-        query, args, _ = self._compile(query)
-        return await self._connection.fetchval(query, *args, column=column)
+        # we are not calling self._connection.fetchval here because
+        # it does not convert all the types, e.g. JSON stays string
+        # instead of an object
+        # see also:
+        # https://github.com/encode/databases/pull/131
+        # https://github.com/encode/databases/pull/132
+        # https://github.com/encode/databases/pull/246
+        row = await self.fetch_one(query)
+        if row is None:
+            return None
+        return row[column]
 
     async def execute(self, query: ClauseElement) -> typing.Any:
         assert self._connection is not None, "Connection is not acquired"

tests/test_databases.py

@@ -155,6 +155,18 @@ async def test_queries(database_url):
             result = await database.fetch_val(query=query)
             assert result == "example1"
 
+            # fetch_val() with no rows
+            query = sqlalchemy.sql.select([notes.c.text]).where(
+                notes.c.text == "impossible"
+            )
+            result = await database.fetch_val(query=query)
+            assert result is None
+
+            # fetch_val() with a different column
+            query = sqlalchemy.sql.select([notes.c.id, notes.c.text])
+            result = await database.fetch_val(query=query, column=1)
+            assert result == "example1"
+
             # row access (needed to maintain test coverage for Record.__getitem__ in postgres backend)
             query = sqlalchemy.sql.select([notes.c.text])
             result = await database.fetch_one(query=query)