Coverage for /opt/conda/envs/apienv/lib/python3.10/site-packages/daiquiri/core/resources/multi_resource_provider.py: 89%

1import contextlib 

2import typing 

3import logging 

4from .resource_provider import ResourceProvider 

5from .resource_provider import ResourceNotAvailable 

6 

7 

8_logger = logging.getLogger(__name__) 

9 

10 

11class MultiResourceProvider(ResourceProvider): 

12 """ 

13 Aggregate resource provider together using priority. 

14 

15 The first added priority have the priority. 

16 """ 

17 

18 def __init__(self): 

19 self.__providers = [] 

20 

21 @property 

22 def providers(self) -> typing.List[ResourceProvider]: 

23 """Sub providers exposed by this resource provider""" 

24 return list(self.__providers) 

25 

26 def __repr__(self): 

27 r = ",".join([repr(p) for p in self.__providers]) 

28 return f"<MultiFileResourceProvider {r}>" 

29 

30 def prepend_provider(self, provider: ResourceProvider): 

31 """Add a new provider with a higher priority than the ones already 

32 registered.""" 

33 self.__providers.insert(0, provider) 

34 

35 def get_resource_path(self, resourcetype: str, resourcename: str) -> str: 

36 """Get the path of a resource from its name from this resource provider.""" 

37 for p in self.__providers: 

38 try: 

39 return p.get_resource_path(resourcetype, resourcename) 

40 except ResourceNotAvailable: 

41 pass 

42 

43 raise ResourceNotAvailable( 

44 f"Resource {resourcetype}/{resourcename} not provided" 

45 ) 

46 

47 def list_resource_names( 

48 self, resourcetype: str, pattern: str = "*" 

49 ) -> typing.List[str]: 

50 """List resource names by pattern from the resource provider.""" 

51 for p in self.__providers: 

52 files = p.list_resource_names(resourcetype, pattern) 

53 if files: 

54 return files 

55 return [] 

56 

57 def _create_resource(self, provider, resource): 

58 return (provider, resource) 

59 

60 def resource(self, resourcetype: str, resourcename: str) -> object: 

61 """Returns a resource descriptor 

62 

63 Arguments: 

64 resourcetype: subdirectory 

65 resourcename: filename 

66 """ 

67 for p in self.__providers: 

68 try: 

69 resource = p.resource(resourcetype, resourcename) 

70 return self._create_resource(p, resource) 

71 except ResourceNotAvailable: 

72 pass 

73 

74 raise ResourceNotAvailable( 

75 f"Resource {resourcetype}/{resourcename} not provided" 

76 ) 

77 

78 @contextlib.contextmanager 

79 def open_resource(self, resource: object, mode="t"): 

80 """Context manager with the resource as a stream 

81 

82 Argument: 

83 resource: Reference to the resource 

84 """ 

85 if not isinstance(resource, tuple): 

86 raise TypeError("Tuple expected") 

87 relative_provider, relative_resource = resource 

88 with relative_provider.open_resource(relative_resource, mode=mode) as f: 

89 yield f