Acquire a connection from the pool.
(self)
| 81 | self._connections: List[Connection] = [] |
| 82 | |
| 83 | async def acquire(self) -> Optional[Connection]: |
| 84 | """Acquire a connection from the pool.""" |
| 85 | if self._connections: |
| 86 | return self._connections.pop() |
| 87 | conn = Connection(self._name) |
| 88 | await conn.connect() |
| 89 | return conn |
| 90 | |
| 91 | def release(self, conn: Connection) -> None: |
| 92 | self._connections.append(conn) |
nothing calls this directly
no test coverage detected