Coverage for python/lsst/resources/proxied.py: 82%

1# This file is part of lsst-resources. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (https://www.lsst.org). 

6# See the COPYRIGHT file at the top-level directory of this distribution 

7# for details of code ownership. 

8# 

9# Use of this source code is governed by a 3-clause BSD-style 

10# license that can be found in the LICENSE file. 

11 

12from __future__ import annotations 

13 

14__all__ = ("ProxiedResourcePath",) 

15 

16import contextlib 

17import dataclasses 

18import logging 

19import re 

20from abc import ABC, abstractmethod 

21from collections.abc import Generator, Iterator 

22from typing import TYPE_CHECKING 

23 

24from ._resourcePath import ResourceHandleProtocol, ResourceInfo, ResourcePath, ResourcePathExpression 

25from .utils import TransactionProtocol 

26 

27try: 

28 import fsspec 

29 from fsspec.spec import AbstractFileSystem 

30except ImportError: 

31 # Hidden from type checkers so that the names above keep the types they 

32 # have when fsspec is installed. 

33 if not TYPE_CHECKING: 

34 fsspec = None 

35 AbstractFileSystem = type 

36 

37 

38log = logging.getLogger(__name__) 

39 

40 

41class ProxiedResourcePath(ABC, ResourcePath): 

42 """URI that is represented internally by another type of URI for file I/O. 

43 

44 For example ``abc://xyz/file.txt`` could be the public URI form but 

45 internally all file access is forwarded to a ``file`` URI. 

46 """ 

47 

48 _proxy: ResourcePath | None = None 

49 

50 @abstractmethod 

51 def _set_proxy(self) -> None: 

52 """Calculate the internal `ResourcePath` corresponding to the public 

53 version. 

54 """ 

55 raise NotImplementedError("Proxy must be configured") 

56 

57 def _get_proxy(self) -> ResourcePath: 

58 """Retrieve the proxied ResourcePath.""" 

59 proxy = self._proxy 

60 if proxy is None: 

61 raise FileNotFoundError(f"Internal error: No proxy ResourcePath available for {self}") 

62 return proxy 

63 

64 def to_fsspec(self) -> tuple[AbstractFileSystem, str]: 

65 try: 

66 proxy = self._get_proxy() 

67 except FileNotFoundError: 

68 raise NotImplementedError(f"No proxy registered for {self}. Resource does not exist.") from None 

69 return proxy.to_fsspec() 

70 

71 def isdir(self) -> bool: 

72 if self.dirLike is None: 

73 try: 

74 proxy = self._get_proxy() 

75 except FileNotFoundError: 

76 return False 

77 self.dirLike = proxy.isdir() 

78 return self.dirLike 

79 

80 def exists(self) -> bool: 

81 try: 

82 proxy = self._get_proxy() 

83 except FileNotFoundError: 

84 # If there is no proxy registered then the resource can not exist. 

85 return False 

86 return proxy.exists() 

87 

88 def remove(self) -> None: 

89 proxy = self._get_proxy() 

90 proxy.remove() 

91 

92 def read(self, size: int = -1) -> bytes: 

93 proxy = self._get_proxy() 

94 return proxy.read(size=size) 

95 

96 @contextlib.contextmanager 

97 def as_local( 

98 self, multithreaded: bool = True, tmpdir: ResourcePathExpression | None = None 

99 ) -> Generator[ResourcePath]: 

100 proxy = self._get_proxy() 

101 with proxy.as_local(multithreaded=multithreaded, tmpdir=tmpdir) as loc: 

102 yield loc 

103 

104 @contextlib.contextmanager 

105 def open( 

106 self, 

107 mode: str = "r", 

108 *, 

109 encoding: str | None = None, 

110 prefer_file_temporary: bool = False, 

111 ) -> Generator[ResourceHandleProtocol]: 

112 proxy = self._get_proxy() 

113 with proxy.open(mode, encoding=encoding, prefer_file_temporary=prefer_file_temporary) as fh: 

114 yield fh 

115 

116 def walk( 

117 self, file_filter: str | re.Pattern | None = None 

118 ) -> Iterator[list | tuple[ResourcePath, list[str], list[str]]]: 

119 try: 

120 proxy = self._get_proxy() 

121 except FileNotFoundError as e: 

122 raise ValueError(str(e)) from None 

123 for proxied_root, dirs, files in proxy.walk(file_filter=file_filter): 

124 # Need to return the directory in the original form and not the 

125 # proxy form. 

126 relative_to_self = proxied_root.path.removeprefix(proxy.path) 

127 root = self.replace(path=self._pathModule.join(self.path, relative_to_self)) 

128 yield root, dirs, files 

129 

130 def size(self) -> int: 

131 proxy = self._get_proxy() 

132 return proxy.size() 

133 

134 def get_info(self) -> ResourceInfo: 

135 proxy = self._get_proxy() 

136 return dataclasses.replace(proxy.get_info(), uri=str(self)) 

137 

138 def write(self, data: bytes, overwrite: bool = True) -> None: 

139 proxy = self._get_proxy() 

140 proxy.write(data, overwrite=overwrite) 

141 

142 def mkdir(self) -> None: 

143 proxy = self._get_proxy() 

144 proxy.mkdir() 

145 

146 def transfer_from( 

147 self, 

148 src: ResourcePath, 

149 transfer: str = "copy", 

150 overwrite: bool = False, 

151 transaction: TransactionProtocol | None = None, 

152 multithreaded: bool = True, 

153 ) -> None: 

154 proxy = self._get_proxy() 

155 proxy.transfer_from( 

156 src, transfer=transfer, overwrite=overwrite, transaction=transaction, multithreaded=multithreaded 

157 )