Use backend native fetch_val() implementation when available

databases/backends/postgres.py

@@ -155,6 +155,13 @@ async def fetch_one(self, query: ClauseElement) -> typing.Optional[typing.Mappin
             return None
         return Record(row, result_columns, self._dialect)
 
+    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)
+
     async def execute(self, query: ClauseElement) -> typing.Any:
         assert self._connection is not None, "Connection is not acquired"
         query, args, result_columns = self._compile(query)

databases/core.py

@@ -251,8 +251,7 @@ async def fetch_val(
     ) -> typing.Any:
         built_query = self._build_query(query, values)
         async with self._query_lock:
-            row = await self._connection.fetch_one(built_query)
-        return None if row is None else row[column]
+            return await self._connection.fetch_val(built_query, column)
 
     async def execute(
         self, query: typing.Union[ClauseElement, str], values: dict = None

databases/interfaces.py

@@ -27,6 +27,12 @@ async def fetch_all(self, query: ClauseElement) -> typing.List[typing.Mapping]:
     async def fetch_one(self, query: ClauseElement) -> typing.Optional[typing.Mapping]:
         raise NotImplementedError()  # pragma: no cover
 
+    async def fetch_val(
+        self, query: ClauseElement, column: typing.Any = 0
+    ) -> typing.Any:
+        row = await self.fetch_one(query)
+        return None if row is None else row[column]
+
     async def execute(self, query: ClauseElement) -> typing.Any:
         raise NotImplementedError()  # pragma: no cover
 

tests/test_databases.py

@@ -155,6 +155,12 @@ async def test_queries(database_url):
             result = await database.fetch_val(query=query)
             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)
+            assert result["text"] == "example1"
+            assert result[0] == "example1"
+
             # iterate()
             query = notes.select()
             iterate_results = []
@@ -206,6 +212,16 @@ async def test_queries_raw(database_url):
             assert result["text"] == "example2"
             assert result["completed"] == False
 
+            # fetch_val()
+            query = "SELECT completed FROM notes WHERE text = :text"
+            result = await database.fetch_val(query=query, values={"text": "example1"})
+            assert result == True
+            query = "SELECT * FROM notes WHERE text = :text"
+            result = await database.fetch_val(
+                query=query, values={"text": "example1"}, column="completed"
+            )
+            assert result == True
+
             # iterate()
             query = "SELECT * FROM notes"
             iterate_results = []