(
self, item: _T, block: bool = True, timeout: Optional[float] = None
)
| 283 | raise Full() from err |
| 284 | |
| 285 | def put( |
| 286 | self, item: _T, block: bool = True, timeout: Optional[float] = None |
| 287 | ) -> None: |
| 288 | if not block: |
| 289 | return self.put_nowait(item) |
| 290 | |
| 291 | try: |
| 292 | if timeout is not None: |
| 293 | self.await_(asyncio.wait_for(self._queue.put(item), timeout)) |
| 294 | else: |
| 295 | self.await_(self._queue.put(item)) |
| 296 | except (asyncio.QueueFull, asyncio.TimeoutError) as err: |
| 297 | raise Full() from err |
| 298 | |
| 299 | def get_nowait(self) -> _T: |
| 300 | try: |
nothing calls this directly
no test coverage detected