(self, max_concurrency: int = 10)
| 170 | """ |
| 171 | |
| 172 | def __init__(self, max_concurrency: int = 10): |
| 173 | self.max_concurrency = max_concurrency |
| 174 | self._semaphore: Optional[asyncio.Semaphore] = None |
| 175 | self.tasks: List[asyncio.Task] = [] |
| 176 | |
| 177 | @property |
| 178 | def semaphore(self) -> asyncio.Semaphore: |
nothing calls this directly
no outgoing calls
no test coverage detected