Coverage for slidge/util/lock.py: 100%

1import asyncio 

2import logging 

3from contextlib import asynccontextmanager 

4from typing import Any, AsyncIterator, Hashable 

5 

6 

7class NamedLockMixin: 

8 def __init__(self, *a: Any, **k: Any) -> None: 

9 super().__init__(*a, **k) 

10 self.__locks = dict[Hashable, asyncio.Lock]() 

11 

12 @asynccontextmanager 

13 async def lock(self, id_: Hashable) -> AsyncIterator[None]: 

14 log.trace("getting %s", id_) # type:ignore 

15 locks = self.__locks 

16 if not locks.get(id_): 

17 locks[id_] = asyncio.Lock() 

18 try: 

19 async with locks[id_]: 

20 log.trace("acquired %s", id_) # type:ignore 

21 yield 

22 finally: 

23 log.trace("releasing %s", id_) # type:ignore 

24 waiters = locks[id_]._waiters 

25 if not waiters: 

26 del locks[id_] 

27 log.trace("erasing %s", id_) # type:ignore 

28 

29 def get_lock(self, id_: Hashable) -> asyncio.Lock | None: 

30 return self.__locks.get(id_) 

31 

32 

33log = logging.getLogger(__name__)