| 990 | |
| 991 | |
| 992 | class StatefulSemaphore: |
| 993 | __slots__ = ("_semaphore", "_max_value", "_acquired_count", "_last_reset") |
| 994 | |
| 995 | """ |
| 996 | StatefulSemaphore is a class that wraps an asyncio.Semaphore and provides additional stateful information. |
| 997 | """ |
| 998 | |
| 999 | def __init__(self, value: int): |
| 1000 | """ |
| 1001 | StatefulSemaphore constructor |
| 1002 | """ |
| 1003 | if value < 0: |
| 1004 | raise ValueError("Value must be non-negative.") |
| 1005 | self._semaphore = asyncio.Semaphore(value) |
| 1006 | self._max_value = value |
| 1007 | self._acquired_count = 0 |
| 1008 | self._last_reset = time.monotonic() |
| 1009 | |
| 1010 | async def acquire(self): |
| 1011 | await self._semaphore.acquire() |
| 1012 | self._acquired_count += 1 |
| 1013 | |
| 1014 | def release(self): |
| 1015 | self._semaphore.release() |
| 1016 | |
| 1017 | self._acquired_count = max(0, self._acquired_count - 1) |
| 1018 | |
| 1019 | def locked(self) -> bool: |
| 1020 | return self._semaphore.locked() |
| 1021 | |
| 1022 | @property |
| 1023 | def available(self) -> int: |
| 1024 | return self._max_value - self._acquired_count |
| 1025 | |
| 1026 | @property |
| 1027 | def acquired(self) -> int: |
| 1028 | return self._acquired_count |
| 1029 | |
| 1030 | @property |
| 1031 | def max_value(self) -> int: |
| 1032 | return self._max_value |
| 1033 | |
| 1034 | @property |
| 1035 | def uptime(self) -> float: |
| 1036 | return time.monotonic() - self._last_reset |
| 1037 | |
| 1038 | def status(self) -> dict: |
| 1039 | return { |
| 1040 | "available": self.available, |
| 1041 | "acquired": self.acquired, |
| 1042 | "max_value": self.max_value, |
| 1043 | "uptime": round(self.uptime, 2), |
| 1044 | } |
| 1045 | |
| 1046 | |
| 1047 | def parse_quantization(value: str): |
no outgoing calls
no test coverage detected