(self, token_manager, concurrency_manager: Optional[ConcurrencyManager] = None)
| 18 | class LoadBalancer: |
| 19 | """Token load balancer with load-aware selection""" |
| 20 | |
| 21 | def __init__( |
| 22 | self, token_manager, concurrency_manager: Optional[ConcurrencyManager] = None |
| 23 | ): |
| 24 | self.token_manager = token_manager |
| 25 | self.concurrency_manager = concurrency_manager |
| 26 | self._image_pending: Dict[int, int] = {} |
| 27 | self._video_pending: Dict[int, int] = {} |
| 28 | self._pending_lock = asyncio.Lock() |
| 29 | self._round_robin_state: Dict[str, Optional[int]] = { |
| 30 | "image": None, |
nothing calls this directly
no outgoing calls
no test coverage detected