Method _getconn_with_check_loop

psycopg_pool/psycopg_pool/pool.py:222–246

(self, deadline: float)

            ) from None

    def _getconn_with_check_loop(self, deadline: float) -> CT:
        attempt: AttemptWithBackoff | None = None

        while True:
            conn = self._getconn_unchecked(deadline - monotonic())
            try:
                self._check_connection(conn)
            except CLIENT_EXCEPTIONS:
                self._putconn(conn, from_getconn=True)
            else:
                logger.info("connection given by %r", self.name)
                return conn

            # Delay further checks to avoid a busy loop, using the same
            # backoff policy used in reconnection attempts.
            now = monotonic()
            if not attempt:
                attempt = AttemptWithBackoff(timeout=deadline - now)
            else:
                attempt.update_delay(now)

            if attempt.time_to_give_up(now):
                raise PoolTimeout()
            else:
                sleep(attempt.delay)

    def _getconn_unchecked(self, timeout: float) -> CT:
        # Critical section: decide here if there's a connection ready