Coverage for src/metador_core/container/provider.py: 97%

1"""Abstract Metador container provider interface.""" 

2from typing import Any, Dict, Generic, Optional, Protocol, Tuple, Type, TypeVar, Union 

3 

4from .wrappers import MetadorContainer, MetadorDriver 

5 

6T = TypeVar("T") 

7 

8 

9class ContainerProxy(Protocol[T]): 

10 """Abstract interface for Metador container providers. 

11 

12 This interface acts like a proxy to access containers by some identifier. 

13 

14 The identifier type parameter T is in the simplest case the Metador 

15 container UUID. In more complex cases, it could be a different unique 

16 identifier with a non-trivial relationship to Metador container UUIDs 

17 (many-to-many). Therefore, T is implementation-specific. 

18 

19 There are many ways to store and organize containers, this interface serves 

20 as the implementation target for generic service components such as 

21 container-centric Flask blueprints, so they can be easier reused in 

22 different backends and services. 

23 

24 Note that only containment and retrieval are possible - on purpose. 

25 Knowing and iterating over all containers in a system is not always possible. 

26 """ 

27 

28 def __contains__(self, key: T) -> bool: 

29 """Return whether a resource key is known to the proxy.""" 

30 # return self.get(key) is not None 

31 

32 def get(self, key: T) -> Optional[MetadorContainer]: 

33 """Get a container instance, if resource key is known to the proxy. 

34 

35 Implement this method in subclasses to support the minimal interface. 

36 """ 

37 

38 def __getitem__(self, key: T) -> T: 

39 """Get a container instance, if resource key is known to the proxy. 

40 

41 Default implementation is in terms of `get`. 

42 """ 

43 if ret := self.get(key): 

44 return ret 

45 raise KeyError(key) 

46 

47 

48ContainerArgs = Tuple[Type[MetadorDriver], Any] 

49"""Pair of (driver class, suitable driver arguments). 

50 

51Must be such that `MetadorContainer(driver(source))` yields a working container. 

52""" 

53 

54 

55class SimpleContainerProvider(Generic[T], ContainerProxy[T]): 

56 """Dict-backed container proxy. 

57 

58 It is a minimal reasonable implementation for the interface that can be 

59 used in small apps and does not depend on the container driver, 

60 thus can support all container interface implementations. 

61 """ 

62 

63 _known: Dict[T, ContainerArgs] 

64 """Mapping from container identifier to MetadorContainer constructor args.""" 

65 

66 def __init__(self): 

67 self._known = {} 

68 

69 def __contains__(self, key: T) -> bool: 

70 return key in self._known 

71 

72 def get(self, key: T) -> Optional[MetadorContainer]: 

73 """Get an open container file to access data and metadata, if it exists.""" 

74 if key not in self._known: 

75 return None 

76 driver, source = self._known[key] 

77 return MetadorContainer(driver(source)) 

78 

79 # ---- 

80 

81 def __delitem__(self, key: T): 

82 del self._known[key] 

83 

84 def __setitem__(self, key: T, value: Union[ContainerArgs, MetadorContainer]): 

85 # NOTE: can't do instance check here, because MetadorContainer is a wrapper itself 

86 # so we check if a container is passed by presence of the .metador attribute 

87 if container_toc := getattr(value, "metador", None): 

88 self._known[key] = (container_toc.driver, container_toc.source) 

89 else: 

90 self._known[key] = value 

91 

92 def keys(self): 

93 return self._known.keys()