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

1import asyncio 

2import logging 

3from collections.abc import AsyncIterator, Hashable 

4from contextlib import asynccontextmanager 

5from typing import Any 

6 

7 

8class NamedLockMixin: 

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

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

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

12 

13 @asynccontextmanager 

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

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

16 locks = self.__locks 

17 if not locks.get(id_): 

18 locks[id_] = asyncio.Lock() 

19 try: 

20 async with locks[id_]: 

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

22 yield 

23 finally: 

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

25 waiters = locks[id_]._waiters 

26 if not waiters: 

27 del locks[id_] 

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

29 

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

31 return self.__locks.get(id_) 

32 

33 

34log = logging.getLogger(__name__)