Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/aiohttp/locks.py: 33%

1import asyncio 

2import collections 

3from typing import Any, Deque, Optional 

4 

5 

6class EventResultOrError: 

7 """Event asyncio lock helper class. 

8 

9 Wraps the Event asyncio lock allowing either to awake the 

10 locked Tasks without any error or raising an exception. 

11 

12 thanks to @vorpalsmith for the simple design. 

13 """ 

14 

15 def __init__(self, loop: asyncio.AbstractEventLoop) -> None: 

16 self._loop = loop 

17 self._exc: Optional[BaseException] = None 

18 self._event = asyncio.Event() 

19 self._waiters: Deque[asyncio.Future[Any]] = collections.deque() 

20 

21 def set(self, exc: Optional[BaseException] = None) -> None: 

22 self._exc = exc 

23 self._event.set() 

24 

25 async def wait(self) -> Any: 

26 waiter = self._loop.create_task(self._event.wait()) 

27 self._waiters.append(waiter) 

28 try: 

29 val = await waiter 

30 finally: 

31 self._waiters.remove(waiter) 

32 

33 if self._exc is not None: 

34 raise self._exc 

35 

36 return val 

37 

38 def cancel(self) -> None: 

39 """Cancel all waiters""" 

40 for waiter in self._waiters: 

41 waiter.cancel()