Coverage for greyhorse / app / resources / scoped.py: 100%

1from __future__ import annotations 

2 

3from collections.abc import Callable 

4 

5from greyhorse.monads.effect import ExitCase 

6from greyhorse.monads.io import IO 

7from greyhorse.monads.resource import Resource 

8from greyhorse.result import Result 

9 

10from .slot import ConsumerSlot, ProviderAccessError, ProviderMutAccessError 

11 

12 

13class SlotBackedScopedProvider[T]: 

14 __slots__ = ('_slot',) 

15 

16 def __init__(self, slot: ConsumerSlot[T]) -> None: 

17 self._slot = slot 

18 

19 def open(self) -> Resource[IO, Result[T, ProviderAccessError]]: 

20 slot = self._slot 

21 

22 def acquire() -> Result[T, ProviderAccessError]: 

23 return slot.try_enter_shared() 

24 

25 def release(result: Result[T, ProviderAccessError], ec: ExitCase) -> None: 

26 if result.is_ok(): 

27 slot.exit_shared() 

28 

29 return Resource[IO, Result[T, ProviderAccessError]].make_case( 

30 acquire=IO(acquire), release=lambda v, ec: IO(lambda: release(v, ec)) 

31 ) 

32 

33 def use[R](self, fn: Callable[[T], R]) -> IO[Result[R, ProviderAccessError]]: 

34 return self.open().use(lambda result: IO.pure(result.map(fn))) 

35 

36 

37class SlotBackedScopedMutProvider[T]: 

38 __slots__ = ('_slot',) 

39 

40 def __init__(self, slot: ConsumerSlot[T]) -> None: 

41 self._slot = slot 

42 

43 def open_mut(self) -> Resource[IO, Result[T, ProviderMutAccessError]]: 

44 slot = self._slot 

45 

46 def acquire() -> Result[T, ProviderMutAccessError]: 

47 return slot.try_enter_mut() 

48 

49 def release(result: Result[T, ProviderMutAccessError], ec: ExitCase) -> None: 

50 if result.is_ok(): 

51 slot.exit_mut() 

52 

53 return Resource[IO, Result[T, ProviderMutAccessError]].make_case( 

54 acquire=IO(acquire), release=lambda v, ec: IO(lambda: release(v, ec)) 

55 ) 

56 

57 def use_mut[R](self, fn: Callable[[T], R]) -> IO[Result[R, ProviderMutAccessError]]: 

58 return self.open_mut().use(lambda result: IO.pure(result.map(fn)))