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

1import typing 

2import contextlib 

3import glob 

4import logging 

5import os.path 

6from .resource_provider import ResourceProvider 

7from .resource_provider import ResourceNotAvailable 

8 

9 

10_logger = logging.getLogger(__name__) 

11 

12 

13class FileResourceProvider(ResourceProvider): 

14 """Provide resource from file directory.""" 

15 

16 def __init__(self, root): 

17 self.__root = os.path.realpath(root) 

18 

19 @property 

20 def root(self): 

21 """Root directory of the resource provider""" 

22 return self.__root 

23 

24 def __repr__(self): 

25 return "<FileResourceProvider {self.__root}>" 

26 

27 def _get_resource_path(self, resourcetype, resourcename): 

28 """Get the resource directory.""" 

29 return os.path.join(self.__root, *resourcetype.split("."), resourcename) 

30 

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

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

33 if resourcename.startswith("/"): 

34 raise ValueError( 

35 "Absolute paths are not supported. Found '%s'" % resourcename 

36 ) 

37 fpath = self._get_resource_path(resourcetype, resourcename) 

38 if not os.path.exists(fpath): 

39 raise ResourceNotAvailable( 

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

41 ) 

42 return fpath 

43 

44 def list_resource_names( 

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

46 ) -> typing.List[str]: 

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

48 files = glob.glob(self._get_resource_path(resourcetype, pattern)) 

49 prefix_size = len(self.__root) + 1 + len(resourcetype) + 1 

50 files = [f[prefix_size:] for f in files] 

51 return files 

52 

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

54 """Returns a resource descriptor 

55 

56 Arguments: 

57 resourcetype: subdirectory 

58 resourcename: filename 

59 """ 

60 return self.get_resource_path(resourcetype, resourcename) 

61 

62 @contextlib.contextmanager 

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

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

65 

66 Argument: 

67 resource: Reference to the resource 

68 """ 

69 _logger.info("Using resource '%s'", resource) 

70 mode = {"t": "rt", "b": "rb", "wt": "wt", "wb": "wb"}[mode] 

71 with open(resource, mode) as f: 

72 yield f