Expose some of Connection methods directly on the Pool object.

Specifically Pool now implements the following methods:

 * Pool.execute()
 * Pool.executemany()
 * Pool.fetch()
 * Pool.fetchrow()
 * Pool.fetchval()

Fixes: #39.

Author: 1st1

asyncpg/pool.py

@@ -306,6 +306,67 @@ async def _async__init__(self):
         self._initialized = True
         return self
 
+    async def execute(self, query: str, *args, timeout: float=None) -> str:
+        """Execute an SQL command (or commands).
+
+        Pool performs this operation using one of its connections.  Other than
+        that, it behaves identically to
+        :meth:`Connection.execute() <connection.Connection.execute>`.
+
+        .. versionadded:: 0.10.0
+        """
+        async with self.acquire() as con:
+            return await con.execute(query, *args, timeout=timeout)
+
+    async def executemany(self, command: str, args, *, timeout: float=None):
+        """Execute an SQL *command* for each sequence of arguments in *args*.
+
+        Pool performs this operation using one of its connections.  Other than
+        that, it behaves identically to
+        :meth:`Connection.executemany() <connection.Connection.executemany>`.
+
+        .. versionadded:: 0.10.0
+        """
+        async with self.acquire() as con:
+            return await con.executemany(command, args, timeout=timeout)
+
+    async def fetch(self, query, *args, timeout=None) -> list:
+        """Run a query and return the results as a list of :class:`Record`.
+
+        Pool performs this operation using one of its connections.  Other than
+        that, it behaves identically to
+        :meth:`Connection.fetch() <connection.Connection.fetch>`.
+
+        .. versionadded:: 0.10.0
+        """
+        async with self.acquire() as con:
+            return await con.fetch(query, *args, timeout=timeout)
+
+    async def fetchval(self, query, *args, column=0, timeout=None):
+        """Run a query and return a value in the first row.
+
+        Pool performs this operation using one of its connections.  Other than
+        that, it behaves identically to
+        :meth:`Connection.fetchval() <connection.Connection.fetchval>`.
+
+        .. versionadded:: 0.10.0
+        """
+        async with self.acquire() as con:
+            return await con.fetchval(
+                query, *args, column=column, timeout=timeout)
+
+    async def fetchrow(self, query, *args, timeout=None):
+        """Run a query and return the first row.
+
+        Pool performs this operation using one of its connections.  Other than
+        that, it behaves identically to
+        :meth:`Connection.fetchrow() <connection.Connection.fetchrow>`.
+
+        .. versionadded:: 0.10.0
+        """
+        async with self.acquire() as con:
+            return await con.fetchrow(query, *args, timeout=timeout)
+
     def acquire(self, *, timeout=None):
         """Acquire a database connection from the pool.
 

tests/test_pool.py

@@ -8,8 +8,9 @@
 import asyncio
 import asyncpg
 import inspect
-import platform
 import os
+import platform
+import random
 import unittest
 
 from asyncpg import _testbase as tb
@@ -378,6 +379,84 @@ async def get_xact_id(con):
                 id3 = await get_xact_id(con)
                 self.assertNotEqual(id2, id3)
 
+    async def test_pool_connection_methods(self):
+        async def test_fetch(pool):
+            i = random.randint(0, 20)
+            await asyncio.sleep(random.random() / 100)
+            r = await pool.fetch('SELECT {}::int'.format(i))
+            self.assertEqual(r, [(i,)])
+            return 1
+
+        async def test_fetchrow(pool):
+            i = random.randint(0, 20)
+            await asyncio.sleep(random.random() / 100)
+            r = await pool.fetchrow('SELECT {}::int'.format(i))
+            self.assertEqual(r, (i,))
+            return 1
+
+        async def test_fetchval(pool):
+            i = random.randint(0, 20)
+            await asyncio.sleep(random.random() / 100)
+            r = await pool.fetchval('SELECT {}::int'.format(i))
+            self.assertEqual(r, i)
+            return 1
+
+        async def test_execute(pool):
+            await asyncio.sleep(random.random() / 100)
+            r = await pool.execute('SELECT generate_series(0, 10)')
+            self.assertEqual(r, 'SELECT {}'.format(11))
+            return 1
+
+        async def test_execute_with_arg(pool):
+            i = random.randint(0, 20)
+            await asyncio.sleep(random.random() / 100)
+            r = await pool.execute('SELECT generate_series(0, $1)', i)
+            self.assertEqual(r, 'SELECT {}'.format(i + 1))
+            return 1
+
+        async def run(N, meth):
+            async with self.create_pool(database='postgres',
+                                        min_size=5, max_size=10) as pool:
+
+                coros = [meth(pool) for _ in range(N)]
+                res = await asyncio.gather(*coros, loop=self.loop)
+                self.assertEqual(res, [1] * N)
+
+        methods = [test_fetch, test_fetchrow, test_fetchval,
+                   test_execute, test_execute_with_arg]
+
+        for method in methods:
+            with self.subTest(method=method.__name__):
+                await run(200, method)
+
+    async def test_pool_connection_execute_many(self):
+        async def worker(pool):
+            await asyncio.sleep(random.random() / 100)
+            await pool.executemany('''
+                INSERT INTO exmany VALUES($1, $2)
+            ''', [
+                ('a', 1), ('b', 2), ('c', 3), ('d', 4)
+            ])
+            return 1
+
+        N = 200
+
+        async with self.create_pool(database='postgres',
+                                    min_size=5, max_size=10) as pool:
+
+            await pool.execute('CREATE TABLE exmany (a text, b int)')
+            try:
+
+                coros = [worker(pool) for _ in range(N)]
+                res = await asyncio.gather(*coros, loop=self.loop)
+                self.assertEqual(res, [1] * N)
+
+                n_rows = await pool.fetchval('SELECT count(*) FROM exmany')
+                self.assertEqual(n_rows, N * 4)
+
+            finally:
+                await pool.execute('DROP TABLE exmany')
+
 
 @unittest.skipIf(os.environ.get('PGHOST'), 'using remote cluster for testing')
 class TestHostStandby(tb.ConnectedTestCase):