Coverage for python/lsst/images/serialization/_reader.py: 93%

1# This file is part of lsst-images. 

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"""User-facing ``open`` reader for incremental, component-wise reads.""" 

12 

13from __future__ import annotations 

14 

15__all__ = ("Reader", "open") 

16 

17from collections.abc import Iterator 

18from contextlib import AbstractContextManager, contextmanager 

19from typing import Any, TypeVar, overload 

20 

21from lsst.resources import ResourcePathExpression 

22 

23from ._backends import backend_for_path 

24from ._common import ArchiveTree, ButlerInfo, MetadataValue 

25from ._input_archive import ArchiveInfo, InputArchive 

26from ._io import public_type_for_schema 

27 

28# This pre-python-3.12 declaration is needed so Sphinx (the 

29# autodoc-typehints plugin) can resolve the ``T`` forward reference in the 

30# stringized annotations; the PEP 695 ``[T]`` parameters below are scoped to 

31# their class/function and are not visible in the module globals. 

32T = TypeVar("T") 

33 

34 

35class Reader[T]: 

36 """A handle to an open ``lsst.images`` file. 

37 

38 Returned by `open`. 

39 Lets the caller pull individual components, or the whole object, out of a 

40 file that is opened once; the underlying archive caches dereferenced 

41 pointers so repeated reads share work. 

42 Valid only inside the ``with`` block that produced it. 

43 """ 

44 

45 def __init__( 

46 self, 

47 archive: InputArchive[Any], 

48 tree: ArchiveTree, 

49 info: ArchiveInfo, 

50 expected_cls: type[T] | None, 

51 ) -> None: 

52 self._archive = archive 

53 self._tree = tree 

54 self._info = info 

55 self._expected_cls = expected_cls 

56 self._closed = False 

57 

58 def _check_open(self) -> None: 

59 if self._closed: 

60 raise RuntimeError("Reader is closed; use it only inside its 'with' block.") 

61 

62 @property 

63 def info(self) -> ArchiveInfo: 

64 """Schema name/version/url and format version for this file.""" 

65 return self._info 

66 

67 @property 

68 def metadata(self) -> dict[str, MetadataValue]: 

69 """Flexible metadata stored with the object.""" 

70 return self._tree.metadata 

71 

72 @property 

73 def butler_info(self) -> ButlerInfo | None: 

74 """Butler dataset info stored with the object, or `None`.""" 

75 return self._tree.butler_info 

76 

77 def get_tree(self) -> ArchiveTree: 

78 """Return the validated on-disk tree for advanced, low-level access. 

79 

80 Most callers want `read` or `get_component` instead; the tree is the 

81 raw deserialization model that those methods build on. 

82 """ 

83 self._check_open() 

84 return self._tree 

85 

86 def get_component(self, name: str, **kwargs: Any) -> Any: 

87 """Deserialize and return a single named component. 

88 

89 Raises `~lsst.images.serialization.InvalidComponentError` for an 

90 unknown component name. 

91 """ 

92 self._check_open() 

93 return self._tree.deserialize_component(name, self._archive, **kwargs) 

94 

95 def read(self, **kwargs: Any) -> T: 

96 """Deserialize and return the whole object.""" 

97 self._check_open() 

98 obj = self._tree.deserialize(self._archive, **kwargs) 

99 if hasattr(obj, "_opaque_metadata"): 

100 obj._opaque_metadata = self._archive.get_opaque_metadata() 

101 if self._expected_cls is not None and not isinstance(obj, self._expected_cls): 101 ↛ 102line 101 didn't jump to line 102 because the condition on line 101 was never true

102 raise TypeError( 

103 f"{self._info.schema_name!r} deserialized to {type(obj).__name__}, " 

104 f"not the requested {self._expected_cls.__name__}." 

105 ) 

106 return obj # type: ignore[return-value] 

107 

108 

109@overload 

110def open[T]( 

111 path: ResourcePathExpression, cls: type[T], *, partial: bool = ..., **backend_kwargs: Any 

112) -> AbstractContextManager[Reader[T]]: ... 

113@overload 

114def open( 114 ↛ exitline 114 didn't return from function 'open' because

115 path: ResourcePathExpression, cls: None = ..., *, partial: bool = ..., **backend_kwargs: Any 

116) -> AbstractContextManager[Reader[Any]]: ... 

117@contextmanager 

118def open(path: ResourcePathExpression, cls=None, *, partial=True, **backend_kwargs: Any) -> Iterator[Reader]: 

119 """Open an ``lsst.images`` file for incremental, component-wise reads. 

120 

121 Dispatches to the appropriate backend by file extension, resolves the 

122 registered in-memory type from the file's schema, and returns a `Reader` 

123 context manager. 

124 

125 Parameters 

126 ---------- 

127 path 

128 File to read; convertible to `lsst.resources.ResourcePath`. 

129 cls 

130 Optional expected in-memory type. 

131 When given, `open` validates that the file's schema resolves to a 

132 subclass of ``cls`` (raising `TypeError` otherwise) and the returned 

133 `Reader` is typed accordingly, so `Reader.read` needs no cast. 

134 partial 

135 Forwarded to the backend ``open_tree``; defaults to `True` (a reader 

136 is for incremental access). 

137 A no-op for the JSON and NDF backends. 

138 **backend_kwargs 

139 Backend-specific open options (e.g. ``page_size`` for FITS). 

140 

141 Raises 

142 ------ 

143 ValueError 

144 If the file extension is not recognized. 

145 ArchiveReadError 

146 If the file's schema is not registered. 

147 TypeError 

148 If ``cls`` is given and the file's schema resolves to an 

149 incompatible type. 

150 """ 

151 backend = backend_for_path(path) 

152 with backend.input_archive.open_tree(path, partial=partial, **backend_kwargs) as ( 

153 archive, 

154 tree, 

155 info, 

156 ): 

157 if cls is not None: 

158 resolved = public_type_for_schema(info.schema_name) 

159 if resolved is not None and not issubclass(resolved, cls): 

160 raise TypeError( 

161 f"{path!r} has schema {info.schema_name!r} (type {resolved.__name__}), " 

162 f"which is not a {cls.__name__}." 

163 ) 

164 reader: Reader[Any] = Reader(archive, tree, info, cls) 

165 try: 

166 yield reader 

167 finally: 

168 reader._closed = True