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

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 contextlib import AbstractContextManager, contextmanager 

18from typing import Any, TypeVar, overload 

19 

20from lsst.resources import ResourcePathExpression 

21 

22from ._backends import backend_for_path 

23from ._common import ArchiveReadError, ArchiveTree, ButlerInfo, MetadataValue 

24from ._input_archive import ArchiveInfo, InputArchive 

25from ._io import class_for_schema, public_type_for_schema 

26 

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

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

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

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

31T = TypeVar("T") 

32 

33 

34class Reader[T]: 

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

36 

37 Returned by `open`. 

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

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

40 pointers so repeated reads share work. 

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

42 """ 

43 

44 def __init__( 

45 self, 

46 archive: InputArchive[Any], 

47 tree: ArchiveTree, 

48 info: ArchiveInfo, 

49 expected_cls: type[T] | None, 

50 ) -> None: 

51 self._archive = archive 

52 self._tree = tree 

53 self._info = info 

54 self._expected_cls = expected_cls 

55 self._closed = False 

56 

57 def _check_open(self) -> None: 

58 if self._closed: 

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

60 

61 @property 

62 def info(self) -> ArchiveInfo: 

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

64 return self._info 

65 

66 @property 

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

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

69 return self._tree.metadata 

70 

71 @property 

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

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

74 return self._tree.butler_info 

75 

76 def get_tree(self) -> ArchiveTree: 

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

78 

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

80 raw deserialization model that those methods build on. 

81 """ 

82 self._check_open() 

83 return self._tree 

84 

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

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

87 

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

89 unknown component name. 

90 """ 

91 self._check_open() 

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

93 

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

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

96 self._check_open() 

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

98 if hasattr(obj, "_opaque_metadata"): 

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

100 if self._expected_cls is not None and not isinstance(obj, self._expected_cls): 

101 raise TypeError( 

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

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

104 ) 

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

106 

107 

108@overload 

109def open[T]( 

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

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

112@overload 

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

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

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

116@contextmanager 

117def open(path, cls=None, *, partial=True, **backend_kwargs): 

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

119 

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

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

122 context manager. 

123 

124 Parameters 

125 ---------- 

126 path 

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

128 cls 

129 Optional expected in-memory type. 

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

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

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

133 partial 

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

135 is for incremental access). 

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

137 **backend_kwargs 

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

139 

140 Raises 

141 ------ 

142 ValueError 

143 If the file extension is not recognized. 

144 ArchiveReadError 

145 If the file's schema is not registered. 

146 TypeError 

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

148 incompatible type. 

149 """ 

150 backend = backend_for_path(path) 

151 info = backend.input_archive.get_basic_info(path) 

152 tree_cls = class_for_schema(info.schema_name) 

153 if tree_cls is None: 

154 raise ArchiveReadError(f"No registered schema {info.schema_name!r}; cannot open {path!r}.") 

155 if cls is not None: 

156 resolved = public_type_for_schema(info.schema_name) 

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

158 raise TypeError( 

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

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

161 ) 

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

163 archive, 

164 tree, 

165 ): 

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

167 try: 

168 yield reader 

169 finally: 

170 reader._closed = True