(self, keys: list[ApiKey])
| 150 | """ |
| 151 | |
| 152 | def __init__(self, keys: list[ApiKey]) -> None: |
| 153 | if not keys: |
| 154 | raise ValueError("ApiKeyPool requires at least one key") |
| 155 | self._keys = list(keys) |
| 156 | self._lock = threading.Lock() |
| 157 | self._condition = threading.Condition(self._lock) |
| 158 | self._rate_limits_hit: int = 0 |
| 159 | self._retry_successes: int = 0 |
| 160 | self._total_requests_served: int = 0 |
| 161 | self._peak_active_requests: int = 0 |
| 162 | |
| 163 | # -- Public API ----------------------------------------------------------- |
| 164 |
nothing calls this directly
no outgoing calls
no test coverage detected