Add fetch_val() interface

Author: lovelydinosaur

README.md

@@ -97,6 +97,10 @@ rows = await database.fetch_all(query=query)
 query = notes.select()
 row = await database.fetch_one(query=query)
 
+# Fetch single value, defaults to `column=0`.
+query = notes.select()
+value = await database.fetch_val(query=query)
+
 # Fetch multiple rows without loading them all into memory at once
 query = notes.select()
 async for row in database.iterate(query=query):

databases/core.py

@@ -101,6 +101,12 @@ async def fetch_one(
         async with self.connection() as connection:
             return await connection.fetch_one(query, values)
 
+    async def fetch_val(
+        self, query: typing.Union[ClauseElement, str], values: dict = None, column: typing.Any = 0
+    ) -> typing.Any:
+        async with self.connection() as connection:
+            return await connection.fetch_val(query, values, column=column)
+
     async def execute(
         self, query: typing.Union[ClauseElement, str], values: dict = None
     ) -> typing.Any:
@@ -175,6 +181,12 @@ async def fetch_one(
     ) -> typing.Any:
         return await self._connection.fetch_one(self._build_query(query, values))
 
+    async def fetch_val(
+        self, query: typing.Union[ClauseElement, str], values: dict = None, column: typing.Any = 0
+    ) -> typing.Any:
+        row = await self._connection.fetch_one(self._build_query(query, values))
+        return row[column]
+
     async def execute(
         self, query: typing.Union[ClauseElement, str], values: dict = None
     ) -> typing.Any:

tests/test_databases.py

@@ -137,6 +137,11 @@ async def test_queries(database_url):
             assert result["text"] == "example1"
             assert result["completed"] == True
 
+            # fetch_val()
+            query = sqlalchemy.sql.select([notes.c.text])
+            result = await database.fetch_val(query=query)
+            assert result == "example1"
+
             # iterate()
             query = notes.select()
             iterate_results = []