Method try_acquire

contrib/multilingual/api_pool.py:239–258

Non-blocking acquire — returns a key immediately or ``None``. Unlike :meth:`acquire`, this never blocks. If a slot is available right now, return the least-loaded key; otherwise return ``None``. Useful in async contexts where blocking would stall the event loop.

(self)

                    self._condition.wait(timeout=wait)

    def try_acquire(self) -> ApiKey | None:
        """Non-blocking acquire — returns a key immediately or ``None``.

        Unlike :meth:`acquire`, this never blocks.  If a slot is available
        right now, return the least-loaded key; otherwise return ``None``.
        Useful in async contexts where blocking would stall the event loop.
        """
        with self._lock:
            self._recover_expired_keys(time.monotonic())
            available = [k for k in self._keys if k.available]
            if not available:
                return None
            key = min(available, key=lambda k: k.active_requests)
            key.active_requests += 1
            key.total_requests += 1
            self._total_requests_served += 1
            _now_active = sum(k.active_requests for k in self._keys)
            if _now_active > self._peak_active_requests:
                self._peak_active_requests = _now_active
            return key

    def release(self, key: ApiKey, *, success: bool = True) -> None:
        """Release a slot on *key* back to the pool.