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

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, abstractproperty 

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 @abstractproperty 

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

53 

54 @abstractproperty 

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

56 

57 @abstractmethod 

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

59 

60 @abstractproperty 

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

62 

63 @abstractmethod 

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

65 

66 @abstractmethod 

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

68 

69 @abstractproperty 

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

71 

72 @abstractmethod 

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

74 

75 @abstractmethod 

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

77 

78 @abstractmethod 

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

80 

81 @abstractmethod 

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

83 pass 

84 

85 @abstractmethod 

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

87 

88 @abstractmethod 

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

90 

91 @abstractmethod 

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

93 

94 @abstractmethod 

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

96 

97 @abstractmethod 

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

99 

100 @abstractmethod 

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

102 

103 @abstractmethod 

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

105 

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

107 

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

109 self, 

110 exc_type: type[BaseException] | None, 

111 exc_val: BaseException | None, 

112 exc_tb: TracebackType | None, 

113 /, 

114 ) -> bool | None: ... 

115 

116 

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

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

119 `~lsst.resources.ResourcePath` subclasses. 

120 

121 Parameters 

122 ---------- 

123 mode : `str` 

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

125 log : `~logging.Logger` 

126 Logger to used when writing messages. 

127 uri : `lsst.resources.ResourcePath` 

128 The URI of the resource being opened. 

129 newline : `str` 

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

131 Defaults to newline. 

132 

133 Notes 

134 ----- 

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

136 corresponding methods in the `io` module. 

137 """ 

138 

139 _closed: CloseStatus 

140 _mode: str 

141 _log: logging.Logger 

142 _newline: U 

143 _uri: ResourcePath 

144 

145 def __init__( 

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

147 ) -> None: 

148 if newline is None: 

149 if "b" in mode: 

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

151 else: 

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

153 else: 

154 self._newline = newline # type: ignore 

155 self._mode = mode 

156 self._log = log 

157 self._uri = uri 

158 

159 @property 

160 def name(self) -> str: 

161 # Use name for compatibility with TextIOWrapper. 

162 return str(self._uri) 

163 

164 @property 

165 def mode(self) -> str: 

166 return self._mode 

167 

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

169 self._closed = CloseStatus.OPEN 

170 return self 

171 

172 def __exit__( 

173 self, 

174 exc_type: type[BaseException] | None, 

175 exc_bal: BaseException | None, 

176 exc_tb: TracebackType | None, 

177 ) -> bool | None: 

178 self.close() 

179 return None