Add Pool methods to determine its min, max, current and idle size

asyncpg/pool.py

@@ -118,6 +118,12 @@ def __init__(self, pool, *, max_queries, setup, max_inactive_time):
         self._timeout = None
         self._generation = None
 
+    def is_connected(self):
+        return self._con is not None and not self._con.is_closed()
+
+    def is_idle(self):
+        return not self._in_use
+
     async def connect(self):
         if self._con is not None:
             raise exceptions.InternalClientError(
@@ -444,6 +450,34 @@ async def _initialize(self):
 
                 await asyncio.gather(*connect_tasks)
 
+    def get_size(self):
+        """Return the current number of connections in this pool.
+
+        .. versionadded:: 0.25.0
+        """
+        return sum(h.is_connected() for h in self._holders)
+
+    def get_min_size(self):
+        """Return the minimum number of connections in this pool.
+
+        .. versionadded:: 0.25.0
+        """
+        return self._minsize
+
+    def get_max_size(self):
+        """Return the maximum allowed number of connections in this pool.
+
+        .. versionadded:: 0.25.0
+        """
+        return self._maxsize
+
+    def get_idle_size(self):
+        """Return the current number of idle connections in this pool.
+
+        .. versionadded:: 0.25.0
+        """
+        return sum(h.is_connected() and h.is_idle() for h in self._holders)
+
     def set_connect_args(self, dsn=None, **connect_kwargs):
         r"""Set the new connection arguments for this pool.
 

tests/test_pool.py

@@ -720,6 +720,27 @@ async def test_pool_handles_inactive_connection_errors(self):
 
         await pool.close()
 
+    async def test_pool_size_and_capacity(self):
+        async with self.create_pool(
+            database='postgres',
+            min_size=2,
+            max_size=3,
+        ) as pool:
+            self.assertEqual(pool.get_min_size(), 2)
+            self.assertEqual(pool.get_max_size(), 3)
+            self.assertEqual(pool.get_size(), 2)
+            self.assertEqual(pool.get_idle_size(), 2)
+
+            async with pool.acquire():
+                self.assertEqual(pool.get_idle_size(), 1)
+
+                async with pool.acquire():
+                    self.assertEqual(pool.get_idle_size(), 0)
+
+                    async with pool.acquire():
+                        self.assertEqual(pool.get_size(), 3)
+                        self.assertEqual(pool.get_idle_size(), 0)
+
     @unittest.skipIf(sys.version_info[:2] < (3, 6), 'no asyncgen support')
     async def test_pool_handles_transaction_exit_in_asyncgen_1(self):
         pool = await self.create_pool(database='postgres',