Coverage for python/lsst/resources/_resourceHandles/_baseResourceHandle.py: 79%

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. 

11from __future__ import annotations 

12 

13from types import TracebackType 

14 

15__all__ = ("BaseResourceHandle", "CloseStatus", "ResourceHandleProtocol") 

16 

17import logging 

18from abc import ABC, abstractmethod 

19from collections.abc import Callable, Iterable 

20from enum import Enum, auto 

21from io import SEEK_SET 

22from typing import TYPE_CHECKING, AnyStr, Generic, Protocol, TypeVar 

23 

24if TYPE_CHECKING: 

25 # Circular import. 

26 from .._resourcePath import ResourcePath 

27 

28S = TypeVar("S", bound="ResourceHandleProtocol") 

29T = TypeVar("T", bound="BaseResourceHandle") 

30U = TypeVar("U", str, bytes) 

31 

32 

33class CloseStatus(Enum): 

34 """Enumerated closed/open status of a file handle, implementation detail 

35 that may be used by BaseResourceHandle children. 

36 """ 

37 

38 OPEN = auto() 

39 CLOSING = auto() 

40 CLOSED = auto() 

41 

42 

43class ResourceHandleProtocol(Protocol, Generic[U]): 

44 """Defines the interface protocol that is compatible with children of 

45 `.BaseResourceHandle`. 

46 

47 Any class that satisfies this protocol can be used in any context where a 

48 `.BaseResourceHandle` is expected. 

49 """ 

50 

51 @property 

52 @abstractmethod 

53 def mode(self) -> str: ... 53 ↛ exitline 53 didn't return from function 'mode' because

54 

55 @property 

56 @abstractmethod 

57 def name(self) -> str: ... 57 ↛ exitline 57 didn't return from function 'name' because

58 

59 @abstractmethod 

60 def close(self) -> None: ... 60 ↛ exitline 60 didn't return from function 'close' because

61 

62 @property 

63 @abstractmethod 

64 def closed(self) -> bool: ... 64 ↛ exitline 64 didn't return from function 'closed' because

65 

66 @abstractmethod 

67 def fileno(self) -> int: ... 67 ↛ exitline 67 didn't return from function 'fileno' because

68 

69 @abstractmethod 

70 def flush(self) -> None: ... 70 ↛ exitline 70 didn't return from function 'flush' because

71 

72 @property 

73 @abstractmethod 

74 def isatty(self) -> bool | Callable[[], bool]: ... 74 ↛ exitline 74 didn't return from function 'isatty' because

75 

76 @abstractmethod 

77 def readable(self) -> bool: ... 77 ↛ exitline 77 didn't return from function 'readable' because

78 

79 @abstractmethod 

80 def readline(self, size: int = -1, /) -> U: ... 80 ↛ exitline 80 didn't return from function 'readline' because

81 

82 @abstractmethod 

83 def readlines(self, hint: int = -1, /) -> Iterable[U]: ... 83 ↛ exitline 83 didn't return from function 'readlines' because

84 

85 @abstractmethod 

86 def seek(self, offset: int, whence: int = SEEK_SET, /) -> int: 

87 pass 

88 

89 @abstractmethod 

90 def seekable(self) -> bool: ... 90 ↛ exitline 90 didn't return from function 'seekable' because

91 

92 @abstractmethod 

93 def tell(self) -> int: ... 93 ↛ exitline 93 didn't return from function 'tell' because

94 

95 @abstractmethod 

96 def truncate(self, size: int | None = None, /) -> int: ... 96 ↛ exitline 96 didn't return from function 'truncate' because

97 

98 @abstractmethod 

99 def writable(self) -> bool: ... 99 ↛ exitline 99 didn't return from function 'writable' because

100 

101 @abstractmethod 

102 def writelines(self, lines: Iterable[U], /) -> None: ... 102 ↛ exitline 102 didn't return from function 'writelines' because

103 

104 @abstractmethod 

105 def read(self, size: int = -1, /) -> U: ... 105 ↛ exitline 105 didn't return from function 'read' because

106 

107 @abstractmethod 

108 def write(self, b: U, /) -> int: ... 108 ↛ exitline 108 didn't return from function 'write' because

109 

110 def __enter__(self: S) -> S: ... 110 ↛ exitline 110 didn't return from function '__enter__' because

111 

112 def __exit__( 112 ↛ exitline 112 didn't return from function '__exit__' because

113 self, 

114 exc_type: type[BaseException] | None, 

115 exc_val: BaseException | None, 

116 exc_tb: TracebackType | None, 

117 /, 

118 ) -> bool | None: ... 

119 

120 

121class BaseResourceHandle(ABC, ResourceHandleProtocol[U]): 

122 """Base class interface for the handle like interface of 

123 `~lsst.resources.ResourcePath` subclasses. 

124 

125 Parameters 

126 ---------- 

127 mode : `str` 

128 Handle modes as described in the python `io` module. 

129 log : `~logging.Logger` 

130 Logger to used when writing messages. 

131 uri : `lsst.resources.ResourcePath` 

132 The URI of the resource being opened. 

133 newline : `str` 

134 When doing multiline operations, break the stream on given character 

135 Defaults to newline. 

136 

137 Notes 

138 ----- 

139 Documentation on the methods of this class line should refer to the 

140 corresponding methods in the `io` module. 

141 """ 

142 

143 _closed: CloseStatus 

144 _mode: str 

145 _log: logging.Logger 

146 _newline: U 

147 _uri: ResourcePath 

148 

149 def __init__( 

150 self, mode: str, log: logging.Logger, uri: ResourcePath, *, newline: AnyStr | None = None 

151 ) -> None: 

152 if newline is None: 152 ↛ 153line 152 didn't jump to line 153 because the condition on line 152 was never true

153 if "b" in mode: 

154 self._newline = b"\n" # type: ignore 

155 else: 

156 self._newline = "\n" # type: ignore 

157 else: 

158 self._newline = newline # type: ignore 

159 self._mode = mode 

160 self._log = log 

161 self._uri = uri 

162 

163 @property 

164 def name(self) -> str: 

165 # Use name for compatibility with TextIOWrapper. 

166 return str(self._uri) 

167 

168 @property 

169 def mode(self) -> str: 

170 return self._mode 

171 

172 def __enter__(self: T) -> T: 

173 self._closed = CloseStatus.OPEN 

174 return self 

175 

176 def __exit__( 

177 self, 

178 exc_type: type[BaseException] | None, 

179 exc_bal: BaseException | None, 

180 exc_tb: TracebackType | None, 

181 ) -> bool | None: 

182 self.close() 

183 return None