Coverage for python/lsst/daf/butler/_formatter.py: 83%
533 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-24 02:19 +0000
1# This file is part of daf_butler.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (http://www.lsst.org).
6# See the COPYRIGHT file at the top-level directory of this distribution
7# for details of code ownership.
8#
9# This software is dual licensed under the GNU General Public License and also
10# under a 3-clause BSD license. Recipients may choose which of these licenses
11# to use; please see the files gpl-3.0.txt and/or bsd_license.txt,
12# respectively. If you choose the GPL option then the following text applies
13# (but note that there is still no warranty even if you opt for BSD instead):
14#
15# This program is free software: you can redistribute it and/or modify
16# it under the terms of the GNU General Public License as published by
17# the Free Software Foundation, either version 3 of the License, or
18# (at your option) any later version.
19#
20# This program is distributed in the hope that it will be useful,
21# but WITHOUT ANY WARRANTY; without even the implied warranty of
22# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
23# GNU General Public License for more details.
24#
25# You should have received a copy of the GNU General Public License
26# along with this program. If not, see <http://www.gnu.org/licenses/>.
28from __future__ import annotations
30__all__ = (
31 "FileIntegrityError",
32 "Formatter",
33 "FormatterFactory",
34 "FormatterNotImplementedError",
35 "FormatterParameter",
36 "FormatterV1inV2",
37 "FormatterV2",
38)
40import contextlib
41import copy
42import logging
43import os
44import zipfile
45from abc import ABCMeta, abstractmethod
46from collections.abc import Callable, Iterator, Mapping, Set
47from typing import TYPE_CHECKING, Any, BinaryIO, ClassVar, TypeAlias, final
49from lsst.resources import ResourceHandleProtocol, ResourcePath
50from lsst.utils.introspection import get_full_type_name
51from lsst.utils.timer import time_this
53from ._config import Config
54from ._config_support import LookupKey, processLookupConfigs
55from ._file_descriptor import FileDescriptor
56from ._location import Location
57from ._rubin.temporary_for_ingest import TemporaryForIngest
58from .dimensions import DataCoordinate, DimensionUniverse
59from .mapping_factory import MappingFactory
61log = logging.getLogger(__name__)
63if TYPE_CHECKING:
64 from ._dataset_provenance import DatasetProvenance
65 from ._dataset_ref import DatasetRef
66 from ._dataset_type import DatasetType
67 from ._storage_class import StorageClass
68 from .datastore.cache_manager import AbstractDatastoreCacheManager
70 # Define a new special type for functions that take "entity"
71 Entity: TypeAlias = DatasetType | DatasetRef | StorageClass | str
74class FileIntegrityError(RuntimeError):
75 """The file metadata is inconsistent with the metadata supplied by
76 the datastore.
77 """
80class FormatterNotImplementedError(NotImplementedError):
81 """Formatter does not implement the specific read or write method
82 that is being requested.
83 """
86class FormatterV2:
87 """Interface for reading and writing datasets using URIs.
89 The formatters are associated with a particular `StorageClass`.
91 Parameters
92 ----------
93 file_descriptor : `FileDescriptor`, optional
94 Identifies the file to read or write, and the associated storage
95 classes and parameter information.
96 ref : `DatasetRef`
97 The dataset associated with this formatter. Should not be a component
98 dataset ref.
99 write_parameters : `dict`, optional
100 Parameters to control how the dataset is serialized.
101 write_recipes : `dict`, optional
102 Detailed write recipes indexed by recipe name.
104 **kwargs
105 Additional arguments that will be ignored but allow for
106 `Formatter` V1 parameters to be given.
108 Notes
109 -----
110 A `FormatterV2` author should not override the default `read` or `write`
111 method. Instead for read the formatter author should implement one or all
112 of `read_from_stream`, `read_from_uri`, or `read_from_local_file`. The
113 method `read_from_uri` will always be attempted first and could be more
114 efficient (since it allows the possibility for a subset of the data file to
115 be accessed remotely when parameters or components are specified) but it
116 will not update the local cache. If the entire contents of the remote file
117 are being accessed (no component or parameters defined) and the dataset
118 would be cached, `read_from_uri` will be called with a local file. If the
119 file is remote and the parameters that have been included are known to be
120 more efficiently handled with a local file, the `read_from_uri` method can
121 return `NotImplemented` to indicate that a local file should be given
122 instead.
124 Similarly for writes, the `write` method can not be subclassed. Instead
125 the formatter author should implement `to_bytes` or `write_local_file`.
126 For local URIs the system will always call `write_local_file` first (which
127 by default will call `to_bytes`) to ensure atomic writes are implemented.
128 For remote URIs with local caching disabled, `to_bytes` will be called
129 first and the remote updated directly. If the dataset should be cached
130 it will always be written locally first.
131 """
133 unsupported_parameters: ClassVar[Set[str] | None] = frozenset()
134 """Set of read parameters not understood by this `Formatter`. An empty set
135 means all parameters are supported. `None` indicates that no parameters
136 are supported. These parameters should match those defined in the storage
137 class definition. (`frozenset`).
138 """
140 supported_write_parameters: ClassVar[Set[str] | None] = None
141 """Parameters understood by this formatter that can be used to control
142 how a dataset is serialized. `None` indicates that no parameters are
143 supported."""
145 default_extension: ClassVar[str | None] = None
146 """Default extension to use when writing a file.
148 Can be `None` if the extension is determined dynamically. Use the
149 `get_write_extension` method to get the actual extension to use.
150 """
152 supported_extensions: ClassVar[Set[str]] = frozenset()
153 """Set of all extensions supported by this formatter.
155 Any extension assigned to the ``default_extension`` property will be
156 automatically included in the list of supported extensions.
157 """
159 can_read_from_uri: ClassVar[bool] = False
160 """Declare whether `read_from_uri` is available to this formatter."""
162 can_read_from_stream: ClassVar[bool] = False
163 """Declare whether `read_from_stream` is available to this formatter."""
165 can_read_from_local_file: ClassVar[bool] = False
166 """Declare whether `read_from_local_file` is available to this
167 formatter."""
169 def __init__(
170 self,
171 file_descriptor: FileDescriptor,
172 *,
173 ref: DatasetRef,
174 write_parameters: Mapping[str, Any] | None = None,
175 write_recipes: Mapping[str, Any] | None = None,
176 # Compatibility parameters. Unused in v2.
177 **kwargs: Any,
178 ):
179 if not isinstance(file_descriptor, FileDescriptor): 179 ↛ 180line 179 didn't jump to line 180 because the condition on line 179 was never true
180 raise TypeError("File descriptor must be a FileDescriptor")
182 self._file_descriptor = file_descriptor
184 if ref.isComponent():
185 # It is a component ref for disassembled composites.
186 ref = ref.makeCompositeRef()
187 self._dataset_ref = ref
189 # Check that the write parameters are allowed
190 if write_parameters:
191 if self.supported_write_parameters is None: 191 ↛ 192line 191 didn't jump to line 192 because the condition on line 191 was never true
192 raise ValueError(
193 f"This formatter does not accept any write parameters. Got: {', '.join(write_parameters)}"
194 )
195 else:
196 given = set(write_parameters)
197 unknown = given - self.supported_write_parameters
198 if unknown: 198 ↛ 199line 198 didn't jump to line 199 because the condition on line 198 was never true
199 s = "s" if len(unknown) != 1 else ""
200 unknownStr = ", ".join(f"'{u}'" for u in unknown)
201 raise ValueError(f"This formatter does not accept parameter{s} {unknownStr}")
203 self._write_parameters = write_parameters
204 self._write_recipes = self.validate_write_recipes(write_recipes)
206 def __str__(self) -> str:
207 return f"{self.name()}@{self.file_descriptor.location.uri}"
209 def __repr__(self) -> str:
210 return f"{self.name()}({self.file_descriptor!r})"
212 @property
213 def file_descriptor(self) -> FileDescriptor:
214 """File descriptor associated with this formatter
215 (`FileDescriptor`).
216 """
217 return self._file_descriptor
219 @property
220 def data_id(self) -> DataCoordinate:
221 """Return Data ID associated with this formatter (`DataCoordinate`)."""
222 return self._dataset_ref.dataId
224 @property
225 def dataset_ref(self) -> DatasetRef:
226 """Return Dataset Ref associated with this formatter (`DatasetRef`)."""
227 return self._dataset_ref
229 @property
230 def write_parameters(self) -> Mapping[str, Any]:
231 """Parameters to use when writing out datasets."""
232 if self._write_parameters is not None:
233 return self._write_parameters
234 return {}
236 @property
237 def write_recipes(self) -> Mapping[str, Any]:
238 """Detailed write Recipes indexed by recipe name."""
239 if self._write_recipes is not None:
240 return self._write_recipes
241 return {}
243 def get_write_extension(self) -> str:
244 """Extension to use when writing a file."""
245 default_extension = self.default_extension
246 extension = default_extension if default_extension is not None else ""
247 return extension
249 def can_accept(self, in_memory_dataset: Any) -> bool:
250 """Indicate whether this formatter can accept the specified
251 storage class directly.
253 Parameters
254 ----------
255 in_memory_dataset : `object`
256 The dataset that is to be written.
258 Returns
259 -------
260 accepts : `bool`
261 If `True` the formatter can write data of this type without
262 requiring datastore to convert it. If `False` the datastore
263 will attempt to convert before writing.
265 Notes
266 -----
267 The base class always returns `False` even if the given type is an
268 instance of the storage class type. This will result in a storage
269 class conversion no-op but also allows mocks with mocked storage
270 classes to work properly.
271 """
272 return False
274 @classmethod
275 def validate_write_recipes(cls, recipes: Mapping[str, Any] | None) -> Mapping[str, Any] | None:
276 """Validate supplied recipes for this formatter.
278 The recipes are supplemented with default values where appropriate.
280 Parameters
281 ----------
282 recipes : `dict`
283 Recipes to validate.
285 Returns
286 -------
287 validated : `dict`
288 Validated recipes.
290 Raises
291 ------
292 RuntimeError
293 Raised if validation fails. The default implementation raises
294 if any recipes are given.
295 """
296 if recipes: 296 ↛ 297line 296 didn't jump to line 297 because the condition on line 296 was never true
297 raise RuntimeError(f"This formatter does not understand these write recipes: {recipes}")
298 return recipes
300 @classmethod
301 def name(cls) -> str:
302 """Return the fully qualified name of the formatter.
304 Returns
305 -------
306 name : `str`
307 Fully-qualified name of formatter class.
308 """
309 return get_full_type_name(cls)
311 def _is_disassembled(self) -> bool:
312 """Return `True` if this formatter is looking at a disassembled
313 component.
314 """
315 return self.file_descriptor.component is not None
317 def _check_resource_size(self, uri: ResourcePath, recorded_size: int, resource_size: int) -> None:
318 """Compare the recorded size with the resource size.
320 The given URI will not be accessed.
321 """
322 if recorded_size >= 0 and resource_size != recorded_size: 322 ↛ 323line 322 didn't jump to line 323 because the condition on line 322 was never true
323 raise FileIntegrityError(
324 "Integrity failure in Datastore. "
325 f"Size of file {uri} ({resource_size}) "
326 f"does not match size recorded in registry of {recorded_size}"
327 )
329 def _get_cache_ref(self) -> DatasetRef:
330 """Get the `DatasetRef` to use for cache look ups.
332 Returns
333 -------
334 ref : `lsst.daf.butler.DatasetRef`
335 The dataset ref to use when looking in the cache.
336 For single-file dataset this will be the dataset ref directly.
337 If this is disassembled we need the component and the component
338 will be in the `FileDescriptor`.
339 """
340 if self.file_descriptor.component is None:
341 cache_ref = self.dataset_ref
342 else:
343 cache_ref = self.dataset_ref.makeComponentRef(self.file_descriptor.component)
344 return cache_ref
346 def _ensure_cache(
347 self, cache_manager: AbstractDatastoreCacheManager | None = None
348 ) -> AbstractDatastoreCacheManager:
349 """Return the cache if given else return a null cache."""
350 if cache_manager is None: 350 ↛ 352line 350 didn't jump to line 352 because the condition on line 350 was never true
351 # Circular import avoidance.
352 from .datastore.cache_manager import DatastoreDisabledCacheManager
354 cache_manager = DatastoreDisabledCacheManager(None, None)
355 return cache_manager
357 def read(
358 self,
359 component: str | None = None,
360 expected_size: int = -1,
361 cache_manager: AbstractDatastoreCacheManager | None = None,
362 ) -> Any:
363 """Read a Dataset.
365 Parameters
366 ----------
367 component : `str`, optional
368 Component to read from the file. Only used if the `StorageClass`
369 for reading differed from the `StorageClass` used to write the
370 file.
371 expected_size : `int`, optional
372 If known, the expected size of the resource to read. This can be
373 used for verification or to decide whether to do a direct read or a
374 file download. ``-1`` indicates the file size is not known.
375 cache_manager : `AbstractDatastoreCacheManager`
376 A cache manager to use to allow a formatter to cache a remote file
377 locally or read a cached file that is already local.
379 Returns
380 -------
381 in_memory_dataset : `object`
382 The requested Dataset.
384 Notes
385 -----
386 This method should not be subclassed. Instead formatter subclasses
387 should re-implement the specific ``read_from_*`` methods as
388 appropriate. Each of these methods has a corresponding class property
389 that must be `True` for the method to be called.
391 The priority for reading is:
393 * `read_from_uri`
394 * `read_from_stream`
395 * `read_from_local_file`
396 * `read_from_uri` (but with a local file)
398 Any of these methods can return `NotImplemented` if there is a desire
399 to skip to the next one in the list. If a dataset is being requested
400 with no component, no parameters, and it should also be added to the
401 local cache, the first two calls will be skipped (unless
402 `read_from_stream` is the only implemented read method) such that a
403 local file will be used.
405 A Formatter can also read a file from within a Zip file if the
406 URI associated with the `FileDescriptor` corresponds to a file with
407 a `.zip` extension and a URI fragment of the form
408 ``zip-path={path_in_zip}``. When reading a file from within a Zip
409 file the priority for reading is:
411 * `read_from_stream`
412 * `read_from_local_file`
413 * `read_from_uri`
415 There are multiple cases that must be handled for reading:
417 For a single file:
419 * No component requested, read the whole file.
420 * Component requested, optionally read the component efficiently,
421 else read the whole file and extract the component.
422 * Derived component requested, read whole file or read relevant
423 component and derive.
425 Disassembled Composite:
427 * The file to read here is the component itself. Formatter only knows
428 about this one component file. Should be no component specified
429 in the ``read`` call but the `FileDescriptor` will know which
430 component this is.
431 * A derived component. The file to read is a component but not the
432 specified component. The caching needs the component from which
433 it's derived.
435 Raises
436 ------
437 lsst.daf.butler.FormatterNotImplementedError
438 Raised if no implementations were found that could read this
439 resource.
440 """
441 # If the file to read is a ZIP file with a fragment requesting
442 # a file within the ZIP file, it is no longer possible to use the
443 # direct read from URI option and the contents of the Zip file must
444 # be extracted.
445 uri = self.file_descriptor.location.uri
446 if uri.fragment and uri.unquoted_fragment.startswith("zip-path="):
447 _, _, path_in_zip = uri.unquoted_fragment.partition("=")
449 # Open the Zip file using ResourcePath.
450 with uri.open("rb") as fd:
451 with zipfile.ZipFile(fd) as zf: # type: ignore
452 if self.can_read_from_stream:
453 with contextlib.closing(zf.open(path_in_zip)) as zip_fd:
454 result = self.read_from_stream(zip_fd, component, expected_size=expected_size)
456 if result is not NotImplemented: 456 ↛ 461line 456 didn't jump to line 461 because the condition on line 456 was always true
457 return result
459 # For now for both URI and local file options we retrieve
460 # the bytes to a temporary local and use that.
461 _, suffix = os.path.splitext(path_in_zip)
462 with ResourcePath.temporary_uri(suffix=suffix) as tmp_uri:
463 tmp_uri.write(zf.read(path_in_zip))
465 if self.can_read_from_local_file:
466 result = self.read_from_local_file(
467 tmp_uri.ospath, component, expected_size=expected_size
468 )
469 if result is not NotImplemented: 469 ↛ 471line 469 didn't jump to line 471 because the condition on line 469 was always true
470 return result
471 if self.can_read_from_uri: 471 ↛ 476line 471 didn't jump to line 476
472 result = self.read_from_uri(tmp_uri, component, expected_size=expected_size)
473 if result is not NotImplemented: 473 ↛ 476line 473 didn't jump to line 476
474 return result
476 raise FormatterNotImplementedError(
477 f"Formatter {self.name()} could not read the file at {uri} using any method."
478 )
480 # If the there are no parameters, no component request, and
481 # the file should be cached, it is better to defer the read_from_uri
482 # to use the local file and populate the cache.
483 prefer_local = False
484 if (
485 component is None
486 and not self.file_descriptor.parameters
487 and self._ensure_cache(cache_manager).should_be_cached(self._get_cache_ref())
488 ):
489 prefer_local = True
491 # First see if the formatter can support direct remote read from
492 # a URI. This can be called later from the local path. If it returns
493 # NotImplemented the formatter decided on its own that local
494 # reads are preferred.
495 if not prefer_local and self.can_read_from_uri:
496 result = self.read_directly_from_possibly_cached_uri(
497 component, expected_size, cache_manager=cache_manager
498 )
499 if result is not NotImplemented: 499 ↛ 506line 499 didn't jump to line 506 because the condition on line 499 was always true
500 return result
502 # Some formatters might want to be able to read directly from
503 # an open file stream. This is preferred over forcing a download
504 # to local file system unless a file read option is available and we
505 # want to store it in the cache because the whole file is being read.
506 if self.can_read_from_stream and not (
507 prefer_local and (self.can_read_from_uri or self.can_read_from_local_file)
508 ):
509 result = self.read_from_possibly_cached_stream(
510 component, expected_size, cache_manager=cache_manager
511 )
512 if result is not NotImplemented: 512 ↛ 516line 512 didn't jump to line 516 because the condition on line 512 was always true
513 return result
515 # Finally, try to read the local file.
516 if self.can_read_from_local_file or self.can_read_from_uri:
517 result = self.read_from_possibly_cached_local_file(
518 component, expected_size, cache_manager=cache_manager
519 )
520 if result is not NotImplemented: 520 ↛ 523line 520 didn't jump to line 523 because the condition on line 520 was always true
521 return result
523 raise FormatterNotImplementedError(
524 f"Formatter {self.name()} could not read the file at {uri} using any method."
525 )
527 def _read_from_possibly_cached_location_no_cache_write(
528 self,
529 callback: Callable[[ResourcePath, str | None, int], Any],
530 component: str | None = None,
531 expected_size: int = -1,
532 *,
533 cache_manager: AbstractDatastoreCacheManager | None = None,
534 ) -> Any:
535 """Read from the cache and call payload without writing to cache."""
536 cache_manager = self._ensure_cache(cache_manager)
538 uri = self.file_descriptor.location.uri
539 cache_ref = self._get_cache_ref()
541 # The component for log messages is either the component requested
542 # explicitly or the component from the file descriptor.
543 log_component = component if component is not None else self.file_descriptor.component
545 with cache_manager.find_in_cache(cache_ref, uri.getExtension()) as cached_file:
546 if cached_file is not None:
547 desired_uri = cached_file
548 msg = f" (cached version of {uri})"
549 else:
550 desired_uri = uri
551 msg = ""
553 if desired_uri.isLocal:
554 # Do not spend the time doing a slow size() call to a remote
555 # resource.
556 self._check_resource_size(desired_uri, expected_size, desired_uri.size())
558 with time_this(
559 log,
560 msg="Reading%s from file handle %s%s with formatter %s",
561 args=(
562 f" component {log_component}" if log_component else "",
563 desired_uri,
564 msg,
565 self.name(),
566 ),
567 ):
568 return callback(desired_uri, component, expected_size)
570 def read_from_possibly_cached_stream(
571 self,
572 component: str | None = None,
573 expected_size: int = -1,
574 *,
575 cache_manager: AbstractDatastoreCacheManager | None = None,
576 ) -> Any:
577 """Read from a stream, checking for possible presence in local cache.
579 Parameters
580 ----------
581 component : `str`, optional
582 Component to read from the file. Only used if the `StorageClass`
583 for reading differed from the `StorageClass` used to write the
584 file.
585 expected_size : `int`, optional
586 If known, the expected size of the resource to read. This can be
587 used for verification or to decide whether to do a direct read or a
588 file download. ``-1`` indicates the file size is not known.
589 cache_manager : `AbstractDatastoreCacheManager`
590 A cache manager to use to allow a formatter to check if there is
591 a copy of the file in the local cache.
593 Returns
594 -------
595 in_memory_dataset : `object` or `NotImplemented`
596 The requested Dataset or an indication that the read mode was
597 not implemented.
599 Notes
600 -----
601 Calls `read_from_stream` but will first check the datastore cache
602 in case the file is present locally. This method will not download
603 a file to the local cache.
604 """
606 def _open_stream(uri: ResourcePath, comp: str | None, size: int = -1) -> Any:
607 with uri.open("rb") as fd:
608 return self.read_from_stream(fd, comp, expected_size=size)
610 return self._read_from_possibly_cached_location_no_cache_write(
611 _open_stream, component, expected_size=expected_size, cache_manager=cache_manager
612 )
614 def read_directly_from_possibly_cached_uri(
615 self,
616 component: str | None = None,
617 expected_size: int = -1,
618 *,
619 cache_manager: AbstractDatastoreCacheManager | None = None,
620 ) -> Any:
621 """Read from arbitrary URI, checking for possible presence in local
622 cache.
624 Parameters
625 ----------
626 component : `str`, optional
627 Component to read from the file. Only used if the `StorageClass`
628 for reading differed from the `StorageClass` used to write the
629 file.
630 expected_size : `int`, optional
631 If known, the expected size of the resource to read. This can be
632 ``-1`` indicates the file size is not known.
633 cache_manager : `AbstractDatastoreCacheManager`
634 A cache manager to use to allow a formatter to check if there is
635 a copy of the file in the local cache.
637 Returns
638 -------
639 in_memory_dataset : `object` or `NotImplemented`
640 The requested Dataset or an indication that the read mode was
641 not implemented.
643 Notes
644 -----
645 This method will first check the datastore cache
646 in case the file is present locally. This method will not cache a
647 remote dataset and will only do a size check for local files to avoid
648 unnecessary round trips to a remote server.
650 The URI will be read by calling `read_from_uri`.
651 """
653 def _open_uri(uri: ResourcePath, comp: str | None, size: int = -1) -> Any:
654 return self.read_from_uri(uri, comp, expected_size=size)
656 return self._read_from_possibly_cached_location_no_cache_write(
657 _open_uri, component, expected_size=expected_size, cache_manager=cache_manager
658 )
660 def read_from_possibly_cached_local_file(
661 self,
662 component: str | None = None,
663 expected_size: int = -1,
664 *,
665 cache_manager: AbstractDatastoreCacheManager | None = None,
666 ) -> Any:
667 """Read a dataset ensuring that a local file is used, checking the
668 cache for it.
670 Parameters
671 ----------
672 component : `str`, optional
673 Component to read from the file. Only used if the `StorageClass`
674 for reading differed from the `StorageClass` used to write the
675 file.
676 expected_size : `int`, optional
677 If known, the expected size of the resource to read. This can be
678 used for verification or to decide whether to do a direct read or a
679 file download. ``-1`` indicates the file size is not known.
680 cache_manager : `AbstractDatastoreCacheManager`
681 A cache manager to use to allow a formatter to cache a remote file
682 locally or read a cached file that is already local.
684 Returns
685 -------
686 in_memory_dataset : `object` or `NotImplemented`
687 The requested Dataset or an indication that the read mode was
688 not implemented.
690 Notes
691 -----
692 The file will be downloaded and cached if it is a remote resource.
693 The file contents will be read using `read_from_local_file` or
694 `read_from_uri`, with preference given to the former.
695 """
696 cache_manager = self._ensure_cache(cache_manager)
697 uri = self.file_descriptor.location.uri
699 # Need to have something we can look up in the cache.
700 cache_ref = self._get_cache_ref()
702 # The component for log messages is either the component requested
703 # explicitly or the component from the file descriptor.
704 log_component = component if component is not None else self.file_descriptor.component
706 result = NotImplemented
708 # Ensure we have a local file.
709 with cache_manager.find_in_cache(cache_ref, uri.getExtension()) as cached_file:
710 if cached_file is not None:
711 msg = f"(via cache read of remote file {uri})"
712 uri = cached_file
713 else:
714 msg = ""
716 # If the file is remote and destined for the cache, download it
717 # directly onto the cache file system. This makes the subsequent
718 # move into the cache an atomic rename rather than a
719 # cross-file-system copy, which is what happens when the default
720 # temporary directory and the cache directory are on different
721 # file systems.
722 should_cache = not uri.isLocal and cache_manager.should_be_cached(cache_ref)
723 download_dir = cache_manager.download_directory if should_cache else None
725 with uri.as_local(tmpdir=download_dir) as local_uri:
726 self._check_resource_size(self.file_descriptor.location.uri, expected_size, local_uri.size())
727 can_be_cached = False
728 if uri != local_uri:
729 # URI was remote and file was downloaded
730 cache_msg = ""
732 if should_cache: 732 ↛ 744line 732 didn't jump to line 744 because the condition on line 732 was always true
733 # In this scenario we want to ask if the downloaded
734 # file should be cached but we should not cache
735 # it until after we've used it (to ensure it can't
736 # be expired whilst we are using it).
737 can_be_cached = True
739 # Say that it is "likely" to be cached because
740 # if the formatter read fails we will not be
741 # caching this file.
742 cache_msg = " and likely cached"
744 msg = f"(via download to local file{cache_msg})"
746 with time_this(
747 log,
748 msg="Reading%s from location %s %s with formatter %s",
749 args=(
750 f" component {log_component}" if log_component else "",
751 uri,
752 msg,
753 self.name(),
754 ),
755 ):
756 if self.can_read_from_local_file:
757 result = self.read_from_local_file(
758 local_uri.ospath, component=component, expected_size=expected_size
759 )
760 if result is NotImplemented and self.can_read_from_uri:
761 # If the direct URI reader was skipped earlier and
762 # there is no explicit local file implementation, pass
763 # in the guaranteed local URI to the generic reader.
764 result = self.read_from_uri(
765 local_uri, component=component, expected_size=expected_size
766 )
768 # File was read successfully so can move to cache.
769 # Also move to cache even if NotImplemented was returned.
770 if can_be_cached:
771 cache_manager.move_to_cache(local_uri, cache_ref)
773 return result
775 def read_from_uri(self, uri: ResourcePath, component: str | None = None, expected_size: int = -1) -> Any:
776 """Read a dataset from a URI that can be local or remote.
778 Parameters
779 ----------
780 uri : `lsst.resources.ResourcePath`
781 URI to use to read the dataset. This URI can be local or remote
782 and can refer to the actual resource or to a locally cached file.
783 component : `str` or `None`, optional
784 The component to be read from the dataset.
785 expected_size : `int`, optional
786 If known, the expected size of the resource to read. This can be
787 ``-1`` indicates the file size is not known.
789 Returns
790 -------
791 in_memory_dataset : `object` or `NotImplemented`
792 The Python object read from the resource or `NotImplemented`.
794 Raises
795 ------
796 lsst.daf.butler.FormatterNotImplementedError
797 Raised if there is no support for direct reads from a, possibly,
798 remote URI.
800 Notes
801 -----
802 This method is only called if the class property
803 ``can_read_from_uri`` is set to `True`.
805 It is possible that a cached local file will be given to this method
806 even if it was originally a remote URI. This can happen if the original
807 write resulted in the file being added to the local cache.
809 If the full file is being read this file will not be added to the
810 local cache. Consider returning `NotImplemented` in
811 this situation, for example if there are no parameters or component
812 specified, and allowing the system to fall back to calling
813 `read_from_local_file` (which will populate the cache if configured
814 to do so).
815 """
816 return NotImplemented
818 def read_from_stream(
819 self, stream: BinaryIO | ResourceHandleProtocol, component: str | None = None, expected_size: int = -1
820 ) -> Any:
821 """Read from an open file descriptor.
823 Parameters
824 ----------
825 stream : `lsst.resources.ResourceHandleProtocol` or \
826 `typing.BinaryIO`
827 File stream to use to read the dataset.
828 component : `str` or `None`, optional
829 The component to be read from the dataset.
830 expected_size : `int`, optional
831 If known, the expected size of the resource to read. This can be
832 ``-1`` indicates the file size is not known.
834 Returns
835 -------
836 in_memory_dataset : `object` or `NotImplemented`
837 The Python object read from the stream or `NotImplemented`.
839 Notes
840 -----
841 Only called if the class property ``can_read_from_stream`` is `True`.
842 """
843 return NotImplemented
845 def read_from_local_file(self, path: str, component: str | None = None, expected_size: int = -1) -> Any:
846 """Read a dataset from a URI guaranteed to refer to the local file
847 system.
849 Parameters
850 ----------
851 path : `str`
852 Path to a local file that should be read.
853 component : `str` or `None`, optional
854 The component to be read from the dataset.
855 expected_size : `int`, optional
856 If known, the expected size of the resource to read. This can be
857 ``-1`` indicates the file size is not known.
859 Returns
860 -------
861 in_memory_dataset : `object` or `NotImplemented`
862 The Python object read from the resource or `NotImplemented`.
864 Raises
865 ------
866 lsst.daf.butler.FormatterNotImplementedError
867 Raised if there is no implementation written to read data
868 from a local file.
870 Notes
871 -----
872 This method will only be called if the class property
873 ``can_read_from_local_file`` is `True` and other options were not
874 used.
875 """
876 return NotImplemented
878 def add_provenance(
879 self, in_memory_dataset: Any, /, *, provenance: DatasetProvenance | None = None
880 ) -> Any:
881 """Add provenance to the dataset.
883 Parameters
884 ----------
885 in_memory_dataset : `object`
886 The dataset to serialize.
887 provenance : `~lsst.daf.butler.DatasetProvenance` or `None`, optional
888 Provenance to attach to dataset.
890 Returns
891 -------
892 dataset_to_write : `object`
893 The dataset to use for serialization. Can be the same object as
894 given.
896 Notes
897 -----
898 The base class implementation returns the given object unchanged.
899 """
900 return in_memory_dataset
902 @final
903 def write(
904 self,
905 in_memory_dataset: Any,
906 /,
907 *,
908 cache_manager: AbstractDatastoreCacheManager | None = None,
909 provenance: DatasetProvenance | None = None,
910 ) -> None:
911 """Write a Dataset.
913 Parameters
914 ----------
915 in_memory_dataset : `object`
916 The Dataset to serialize.
917 cache_manager : `AbstractDatastoreCacheManager`
918 A cache manager to use to allow a formatter to cache the written
919 file.
920 provenance : `DatasetProvenance` | `None`, optional
921 Provenance to attach to the file being written.
923 Returns
924 -------
925 None
927 Raises
928 ------
929 lsst.daf.butler.FormatterNotImplementedError
930 Raised if the formatter subclass has not implemented
931 `write_local_file` and `to_bytes` was not called.
932 Exception
933 Raised if there is an error serializing the dataset to disk.
935 Notes
936 -----
937 The intent is for subclasses to implement either `to_bytes` or
938 `write_local_file` or both and not to subclass this method.
939 """
940 # Ensure we are using the correct file extension.
941 uri = self.file_descriptor.location.uri.updatedExtension(self.get_write_extension())
943 # Attach any provenance to the dataset. This could involve returning
944 # a different object.
945 in_memory_dataset = self.add_provenance(in_memory_dataset, provenance=provenance)
947 written = self.write_direct(in_memory_dataset, uri, cache_manager)
948 if not written: 948 ↛ exitline 948 didn't return from function 'write' because the condition on line 948 was always true
949 self.write_locally_then_move(in_memory_dataset, uri, cache_manager)
951 def write_direct(
952 self,
953 in_memory_dataset: Any,
954 uri: ResourcePath,
955 cache_manager: AbstractDatastoreCacheManager | None = None,
956 ) -> bool:
957 """Serialize and write directly to final location.
959 Parameters
960 ----------
961 in_memory_dataset : `object`
962 The Dataset to serialize.
963 uri : `lsst.resources.ResourcePath`
964 URI to use when writing the serialized dataset.
965 cache_manager : `AbstractDatastoreCacheManager`
966 A cache manager to use to allow a formatter to cache the written
967 file.
969 Returns
970 -------
971 written : `bool`
972 Flag to indicate whether the direct write did happen.
974 Raises
975 ------
976 Exception
977 Raised if there was a failure from serializing to bytes that
978 was not `~lsst.daf.butler.FormatterNotImplementedError`.
980 Notes
981 -----
982 This method will call `to_bytes` to serialize the in-memory dataset
983 and then will call the `~lsst.resources.ResourcePath.write` method
984 directly.
986 If the dataset should be cached or is local the file will not be
987 written and the method will return `False`. This is because local URIs
988 should be written to a temporary file name and then renamed to allow
989 atomic writes. That path is handled by `write_locally_then_move`
990 through `write_local_file`) and is preferred over this method being
991 subclassed and the atomic write re-implemented.
992 """
993 cache_manager = self._ensure_cache(cache_manager)
995 # For remote URIs some datasets can be serialized directly
996 # to bytes and sent to the remote datastore without writing a
997 # file. If the dataset is intended to be saved to the cache
998 # a file is always written and direct write to the remote
999 # datastore is bypassed.
1000 data_written = False
1001 if not uri.isLocal and not cache_manager.should_be_cached(self._get_cache_ref()): 1001 ↛ 1003line 1001 didn't jump to line 1003 because the condition on line 1001 was never true
1002 # Remote URI that is not cached so can write directly.
1003 try:
1004 serialized_dataset = self.to_bytes(in_memory_dataset)
1005 except FormatterNotImplementedError:
1006 # Fallback to the file writing option.
1007 pass
1008 except Exception as e:
1009 e.add_note(
1010 f"Failed to serialize dataset {self.dataset_ref} of "
1011 f"type {get_full_type_name(in_memory_dataset)} to bytes."
1012 )
1013 raise
1014 else:
1015 log.debug("Writing bytes directly to %s", uri)
1016 uri.write(serialized_dataset, overwrite=True)
1017 log.debug("Successfully wrote bytes directly to %s", uri)
1018 data_written = True
1019 return data_written
1021 def write_locally_then_move(
1022 self,
1023 in_memory_dataset: Any,
1024 uri: ResourcePath,
1025 cache_manager: AbstractDatastoreCacheManager | None = None,
1026 ) -> None:
1027 """Write file to file system and then move to final location.
1029 Parameters
1030 ----------
1031 in_memory_dataset : `object`
1032 The Dataset to serialize.
1033 uri : `lsst.resources.ResourcePath`
1034 URI to use when writing the serialized dataset.
1035 cache_manager : `AbstractDatastoreCacheManager`
1036 A cache manager to use to allow a formatter to cache the written
1037 file.
1039 Raises
1040 ------
1041 lsst.daf.butler.FormatterNotImplementedError
1042 Raised if the formatter subclass has not implemented
1043 `write_local_file`.
1044 Exception
1045 Raised if there is an error serializing the dataset to disk.
1046 """
1047 cache_manager = self._ensure_cache(cache_manager)
1049 with TemporaryForIngest.make_path(uri) as temporary_uri:
1050 # Need to configure the formatter to write to a different
1051 # location and that needs us to overwrite internals
1052 log.debug("Writing dataset to temporary location at %s", temporary_uri)
1054 # Assumes that if write_local_file is not subclassed that
1055 # to_bytes will be called using the base class definition.
1056 try:
1057 self.write_local_file(in_memory_dataset, temporary_uri)
1058 except Exception as e:
1059 e.add_note(
1060 f"Failed to serialize dataset {self.dataset_ref} of type"
1061 f" {get_full_type_name(in_memory_dataset)} to "
1062 f"temporary location {temporary_uri}."
1063 )
1064 raise
1066 # Use move for a local file since that becomes an efficient
1067 # os.rename. For remote resources we use copy to allow the
1068 # file to be cached afterwards.
1069 transfer = "move" if uri.isLocal else "copy"
1071 uri.transfer_from(temporary_uri, transfer=transfer, overwrite=True)
1073 if transfer == "copy":
1074 # Cache if required
1075 cache_manager.move_to_cache(temporary_uri, self._get_cache_ref())
1077 log.debug("Successfully wrote dataset to %s via a temporary file.", uri)
1079 def write_local_file(self, in_memory_dataset: Any, uri: ResourcePath) -> None:
1080 """Serialize the in-memory dataset to a local file.
1082 Parameters
1083 ----------
1084 in_memory_dataset : `object`
1085 The Python object to serialize.
1086 uri : `~lsst.resources.ResourcePath`
1087 The URI to use when writing the file.
1089 Notes
1090 -----
1091 By default this method will attempt to call `to_bytes` and then
1092 write these bytes to the file.
1094 Raises
1095 ------
1096 lsst.daf.butler.FormatterNotImplementedError
1097 Raised if the formatter subclass has not implemented this method
1098 or has failed to implement the `to_bytes` method.
1099 """
1100 log.debug("Writing bytes directly to %s", uri)
1101 uri.write(self.to_bytes(in_memory_dataset))
1102 log.debug("Successfully wrote bytes directly to %s", uri)
1104 def to_bytes(self, in_memory_dataset: Any) -> bytes:
1105 """Serialize the in-memory dataset to bytes.
1107 Parameters
1108 ----------
1109 in_memory_dataset : `object`
1110 The Python object to serialize.
1112 Returns
1113 -------
1114 serialized_dataset : `bytes`
1115 Bytes representing the serialized dataset.
1117 Raises
1118 ------
1119 lsst.daf.butler.FormatterNotImplementedError
1120 Raised if the formatter has not implemented the method. This will
1121 not cause a problem if `write_local_file` has been implemented.
1122 """
1123 raise FormatterNotImplementedError(
1124 f"This formatter can not convert {get_full_type_name(in_memory_dataset)} directly to bytes."
1125 )
1127 def make_updated_location(self, location: Location) -> Location:
1128 """Return a new `Location` updated with this formatter's extension.
1130 Parameters
1131 ----------
1132 location : `Location`
1133 The location to update.
1135 Returns
1136 -------
1137 updated : `Location`
1138 A new `Location` with a new file extension applied.
1139 """
1140 location = location.clone()
1141 # If the extension is "" the extension will be removed.
1142 location.updateExtension(self.get_write_extension())
1143 return location
1145 @classmethod
1146 def validate_extension(cls, location: Location) -> None:
1147 """Check the extension of the provided location for compatibility.
1149 Parameters
1150 ----------
1151 location : `Location`
1152 Location from which to extract a file extension.
1154 Returns
1155 -------
1156 None
1158 Raises
1159 ------
1160 ValueError
1161 Raised if the formatter does not understand this extension.
1162 """
1163 supported = set(cls.supported_extensions)
1164 default = cls.default_extension # type: ignore
1166 # If extension is implemented as an instance property it won't return
1167 # a string when called as a class property. Assume that
1168 # the supported extensions class property is complete.
1169 if default is not None and isinstance(default, str): 1169 ↛ 1173line 1169 didn't jump to line 1173 because the condition on line 1169 was always true
1170 supported.add(default)
1172 # Get the file name from the uri
1173 file = location.uri.basename()
1175 # Check that this file name ends with one of the supported extensions.
1176 # This is less prone to confusion than asking the location for
1177 # its extension and then doing a set comparison
1178 for ext in supported: 1178 ↛ 1182line 1178 didn't jump to line 1182 because the loop on line 1178 didn't complete
1179 if file.endswith(ext):
1180 return
1182 raise ValueError(
1183 f"Extension '{location.getExtension()}' on '{location}' "
1184 f"is not supported by Formatter '{cls.__name__}' (supports: {supported})"
1185 )
1187 def predict_path(self) -> str:
1188 """Return the path that would be returned by write.
1190 Does not write any data file.
1192 Uses the `FileDescriptor` associated with the instance.
1194 Returns
1195 -------
1196 path : `str`
1197 Path within datastore that would be associated with the location
1198 stored in this `Formatter`.
1199 """
1200 updated = self.make_updated_location(self.file_descriptor.location)
1201 return updated.pathInStore.path
1203 def segregate_parameters(self, parameters: dict[str, Any] | None = None) -> tuple[dict, dict]:
1204 """Segregate the supplied parameters.
1206 This splits the parameters into those understood by the
1207 formatter and those not understood by the formatter.
1209 Any unsupported parameters are assumed to be usable by associated
1210 assemblers.
1212 Parameters
1213 ----------
1214 parameters : `dict`, optional
1215 Parameters with values that have been supplied by the caller
1216 and which might be relevant for the formatter. If `None`
1217 parameters will be read from the registered `FileDescriptor`.
1219 Returns
1220 -------
1221 supported : `dict`
1222 Those parameters supported by this formatter.
1223 unsupported : `dict`
1224 Those parameters not supported by this formatter.
1225 """
1226 if parameters is None: 1226 ↛ 1229line 1226 didn't jump to line 1229 because the condition on line 1226 was always true
1227 parameters = self.file_descriptor.parameters
1229 if parameters is None:
1230 return {}, {}
1232 if self.unsupported_parameters is None:
1233 # Support none of the parameters
1234 return {}, parameters.copy()
1236 # Start by assuming all are supported
1237 supported = parameters.copy()
1238 unsupported = {}
1240 # And remove any we know are not supported
1241 for p in set(supported):
1242 if p in self.unsupported_parameters: 1242 ↛ 1243line 1242 didn't jump to line 1243 because the condition on line 1242 was never true
1243 unsupported[p] = supported.pop(p)
1245 return supported, unsupported
1248class Formatter(metaclass=ABCMeta):
1249 """Interface for reading and writing Datasets.
1251 The formatters are associated with a particular `StorageClass`.
1253 Parameters
1254 ----------
1255 fileDescriptor : `FileDescriptor`, optional
1256 Identifies the file to read or write, and the associated storage
1257 classes and parameter information.
1258 dataId : `DataCoordinate`
1259 Data ID associated with this formatter.
1260 writeParameters : `dict`, optional
1261 Parameters to control how the dataset is serialized.
1262 writeRecipes : `dict`, optional
1263 Detailed write recipes indexed by recipe name.
1264 **kwargs
1265 Additional parameters that can allow parameters
1266 from `FormatterV2` to be provided.
1268 Notes
1269 -----
1270 All Formatter subclasses should share the base class's constructor
1271 signature.
1272 """
1274 # Now assuming that Formatter v1 can only refer to files so can add
1275 # this property for compatibility with v2.
1276 extension: str | None = None
1277 """Default file extension to use for writing files. None means that no
1278 modifications will be made to the supplied file extension. (`str`)"""
1280 unsupportedParameters: ClassVar[Set[str] | None] = frozenset()
1281 """Set of read parameters not understood by this `Formatter`. An empty set
1282 means all parameters are supported. `None` indicates that no parameters
1283 are supported. These parameters should match those defined in the storage
1284 class definition. (`frozenset`).
1285 """
1287 supportedWriteParameters: ClassVar[Set[str] | None] = None
1288 """Parameters understood by this formatter that can be used to control
1289 how a dataset is serialized. `None` indicates that no parameters are
1290 supported."""
1292 supportedExtensions: ClassVar[Set[str]] = frozenset()
1293 """Set of all extensions supported by this formatter.
1295 Only expected to be populated by Formatters that write files. Any extension
1296 assigned to the ``extension`` property will be automatically included in
1297 the list of supported extensions."""
1299 def __init__(
1300 self,
1301 fileDescriptor: FileDescriptor,
1302 *,
1303 dataId: DataCoordinate | None = None,
1304 writeParameters: Mapping[str, Any] | None = None,
1305 writeRecipes: Mapping[str, Any] | None = None,
1306 # Allow FormatterV2 parameters to be dropped.
1307 **kwargs: Any,
1308 ):
1309 if not isinstance(fileDescriptor, FileDescriptor): 1309 ↛ 1310line 1309 didn't jump to line 1310 because the condition on line 1309 was never true
1310 raise TypeError("File descriptor must be a FileDescriptor")
1311 self._fileDescriptor = fileDescriptor
1313 if dataId is None:
1314 raise RuntimeError("dataId is now required for formatter initialization")
1315 if not isinstance(dataId, DataCoordinate): 1315 ↛ 1316line 1315 didn't jump to line 1316 because the condition on line 1315 was never true
1316 raise TypeError(f"DataId is required to be a DataCoordinate but got {type(dataId)}.")
1317 self._dataId = dataId
1319 # V2 compatibility.
1320 if writeParameters is None: 1320 ↛ 1322line 1320 didn't jump to line 1322 because the condition on line 1320 was always true
1321 writeParameters = kwargs.get("write_parameters")
1322 if writeRecipes is None: 1322 ↛ 1326line 1322 didn't jump to line 1326 because the condition on line 1322 was always true
1323 writeRecipes = kwargs.get("write_recipes")
1325 # Check that the write parameters are allowed
1326 if writeParameters:
1327 if self.supportedWriteParameters is None:
1328 raise ValueError(
1329 f"This formatter does not accept any write parameters. Got: {', '.join(writeParameters)}"
1330 )
1331 else:
1332 given = set(writeParameters)
1333 unknown = given - self.supportedWriteParameters
1334 if unknown:
1335 s = "s" if len(unknown) != 1 else ""
1336 unknownStr = ", ".join(f"'{u}'" for u in unknown)
1337 raise ValueError(f"This formatter does not accept parameter{s} {unknownStr}")
1339 self._writeParameters = writeParameters
1340 self._writeRecipes = self.validate_write_recipes(writeRecipes)
1342 def __str__(self) -> str:
1343 return f"{self.name()}@{self.fileDescriptor.location.path}"
1345 def __repr__(self) -> str:
1346 return f"{self.name()}({self.fileDescriptor!r})"
1348 @property
1349 def fileDescriptor(self) -> FileDescriptor:
1350 """File descriptor associated with this formatter
1351 (`FileDescriptor`).
1352 """
1353 return self._fileDescriptor
1355 @property
1356 def dataId(self) -> DataCoordinate:
1357 """Return Data ID associated with this formatter (`DataCoordinate`)."""
1358 return self._dataId
1360 @property
1361 def writeParameters(self) -> Mapping[str, Any]:
1362 """Parameters to use when writing out datasets."""
1363 if self._writeParameters is not None:
1364 return self._writeParameters
1365 return {}
1367 @property
1368 def writeRecipes(self) -> Mapping[str, Any]:
1369 """Detailed write Recipes indexed by recipe name."""
1370 if self._writeRecipes is not None:
1371 return self._writeRecipes
1372 return {}
1374 def can_accept(self, in_memory_dataset: Any) -> bool:
1375 """Indicate whether this formatter can accept the specified
1376 storage class directly.
1378 Parameters
1379 ----------
1380 in_memory_dataset : `object`
1381 The dataset that is to be written.
1383 Returns
1384 -------
1385 accepts : `bool`
1386 If `True` the formatter can write data of this type without
1387 requiring datastore to convert it. If `False` the datastore
1388 will attempt to convert before writing.
1390 Notes
1391 -----
1392 The base class checks that the given python type matches
1393 the python type specified for this formatter when
1394 constructed.
1395 """
1396 return isinstance(in_memory_dataset, self.file_descriptor.storageClass.pytype)
1398 @classmethod
1399 def validateWriteRecipes(cls, recipes: Mapping[str, Any] | None) -> Mapping[str, Any] | None:
1400 """Validate supplied recipes for this formatter.
1402 The recipes are supplemented with default values where appropriate.
1404 Parameters
1405 ----------
1406 recipes : `dict`
1407 Recipes to validate.
1409 Returns
1410 -------
1411 validated : `dict`
1412 Validated recipes.
1414 Raises
1415 ------
1416 RuntimeError
1417 Raised if validation fails. The default implementation raises
1418 if any recipes are given.
1419 """
1420 if recipes: 1420 ↛ 1421line 1420 didn't jump to line 1421 because the condition on line 1420 was never true
1421 raise RuntimeError(f"This formatter does not understand these writeRecipes: {recipes}")
1422 return recipes
1424 @classmethod
1425 def validate_write_recipes(cls, recipes: Mapping[str, Any] | None) -> Mapping[str, Any] | None:
1426 return cls.validateWriteRecipes(recipes)
1428 @classmethod
1429 def name(cls) -> str:
1430 """Return the fully qualified name of the formatter.
1432 Returns
1433 -------
1434 name : `str`
1435 Fully-qualified name of formatter class.
1436 """
1437 return get_full_type_name(cls)
1439 @abstractmethod
1440 def read(self, component: str | None = None) -> Any:
1441 """Read a Dataset.
1443 Parameters
1444 ----------
1445 component : `str`, optional
1446 Component to read from the file. Only used if the `StorageClass`
1447 for reading differed from the `StorageClass` used to write the
1448 file.
1450 Returns
1451 -------
1452 inMemoryDataset : `object`
1453 The requested Dataset.
1454 """
1455 raise FormatterNotImplementedError("Type does not support reading")
1457 @abstractmethod
1458 def write(self, inMemoryDataset: Any) -> None:
1459 """Write a Dataset.
1461 Parameters
1462 ----------
1463 inMemoryDataset : `object`
1464 The Dataset to store.
1465 """
1466 raise FormatterNotImplementedError("Type does not support writing")
1468 @classmethod
1469 def can_read_bytes(cls) -> bool:
1470 """Indicate if this formatter can format from bytes.
1472 Returns
1473 -------
1474 can : `bool`
1475 `True` if the `fromBytes` method is implemented.
1476 """
1477 # We have no property to read so instead try to format from a byte
1478 # and see what happens
1479 try:
1480 # We know the arguments are incompatible
1481 cls.fromBytes(cls, b"") # type: ignore
1482 except FormatterNotImplementedError:
1483 return False
1484 except Exception:
1485 # There will be problems with the bytes we are supplying so ignore
1486 pass
1487 return True
1489 def fromBytes(self, serializedDataset: bytes, component: str | None = None) -> object:
1490 """Read serialized data into a Dataset or its component.
1492 Parameters
1493 ----------
1494 serializedDataset : `bytes`
1495 Bytes object to unserialize.
1496 component : `str`, optional
1497 Component to read from the Dataset. Only used if the `StorageClass`
1498 for reading differed from the `StorageClass` used to write the
1499 file.
1501 Returns
1502 -------
1503 inMemoryDataset : `object`
1504 The requested data as a Python object. The type of object
1505 is controlled by the specific formatter.
1506 """
1507 raise FormatterNotImplementedError("Type does not support reading from bytes.")
1509 def toBytes(self, inMemoryDataset: Any) -> bytes:
1510 """Serialize the Dataset to bytes based on formatter.
1512 Parameters
1513 ----------
1514 inMemoryDataset : `object`
1515 The Python object to serialize.
1517 Returns
1518 -------
1519 serializedDataset : `bytes`
1520 Bytes representing the serialized dataset.
1521 """
1522 raise FormatterNotImplementedError("Type does not support writing to bytes.")
1524 @contextlib.contextmanager
1525 def _updateLocation(self, location: Location | None) -> Iterator[Location]:
1526 """Temporarily replace the location associated with this formatter.
1528 Parameters
1529 ----------
1530 location : `Location`
1531 New location to use for this formatter. If `None` the
1532 formatter will not change but it will still return
1533 the old location. This allows it to be used in a code
1534 path where the location may not need to be updated
1535 but the with block is still convenient.
1537 Yields
1538 ------
1539 old : `Location`
1540 The old location that will be restored.
1542 Notes
1543 -----
1544 This is an internal method that should be used with care.
1545 It may change in the future. Should be used as a context
1546 manager to restore the location when the temporary is no
1547 longer required.
1548 """
1549 old = self._fileDescriptor.location
1550 try:
1551 if location is not None: 1551 ↛ 1553line 1551 didn't jump to line 1553 because the condition on line 1551 was always true
1552 self._fileDescriptor.location = location
1553 yield old
1554 finally:
1555 if location is not None: 1555 ↛ exitline 1555 didn't return from function '_updateLocation' because the condition on line 1555 was always true
1556 self._fileDescriptor.location = old
1558 def make_updated_location(self, location: Location) -> Location:
1559 """Return a new `Location` updated with this formatter's extension.
1561 Parameters
1562 ----------
1563 location : `Location`
1564 The location to update.
1566 Returns
1567 -------
1568 updated : `Location`
1569 A new `Location` with a new file extension applied.
1571 Raises
1572 ------
1573 NotImplementedError
1574 Raised if there is no ``extension`` attribute associated with
1575 this formatter.
1577 Notes
1578 -----
1579 This method is available to all Formatters but might not be
1580 implemented by all formatters. It requires that a formatter set
1581 an ``extension`` attribute containing the file extension used when
1582 writing files. If ``extension`` is `None` the supplied file will
1583 not be updated. Not all formatters write files so this is not
1584 defined in the base class.
1585 """
1586 location = location.clone()
1587 try:
1588 # We are deliberately allowing extension to be undefined by
1589 # default in the base class and mypy complains.
1590 location.updateExtension(self.extension) # type:ignore
1591 except AttributeError:
1592 raise NotImplementedError("No file extension registered with this formatter") from None
1593 return location
1595 @classmethod
1596 def validate_extension(cls, location: Location) -> None:
1597 """Check the extension of the provided location for compatibility.
1599 Parameters
1600 ----------
1601 location : `Location`
1602 Location from which to extract a file extension.
1604 Returns
1605 -------
1606 None
1608 Raises
1609 ------
1610 NotImplementedError
1611 Raised if file extensions are a concept not understood by this
1612 formatter.
1613 ValueError
1614 Raised if the formatter does not understand this extension.
1616 Notes
1617 -----
1618 This method is available to all Formatters but might not be
1619 implemented by all formatters. It requires that a formatter set
1620 an ``extension`` attribute containing the file extension used when
1621 writing files. If ``extension`` is `None` only the set of supported
1622 extensions will be examined.
1623 """
1624 supported = set(cls.supportedExtensions)
1626 try:
1627 # We are deliberately allowing extension to be undefined by
1628 # default in the base class and mypy complains.
1629 default = cls.extension # type: ignore
1630 except AttributeError:
1631 raise NotImplementedError("No file extension registered with this formatter") from None
1633 # If extension is implemented as an instance property it won't return
1634 # a string when called as a class property. Assume that
1635 # the supported extensions class property is complete.
1636 if default is not None and isinstance(default, str): 1636 ↛ 1640line 1636 didn't jump to line 1640 because the condition on line 1636 was always true
1637 supported.add(default)
1639 # Get the file name from the uri
1640 file = location.uri.basename()
1642 # Check that this file name ends with one of the supported extensions.
1643 # This is less prone to confusion than asking the location for
1644 # its extension and then doing a set comparison
1645 for ext in supported:
1646 if file.endswith(ext):
1647 return
1649 raise ValueError(
1650 f"Extension '{location.getExtension()}' on '{location}' "
1651 f"is not supported by Formatter '{cls.__name__}' (supports: {supported})"
1652 )
1654 def predict_path(self) -> str:
1655 """Return the path that would be returned by write.
1657 Does not write any data file.
1659 Uses the `FileDescriptor` associated with the instance.
1661 Returns
1662 -------
1663 path : `str`
1664 Path within datastore that would be associated with the location
1665 stored in this `Formatter`.
1666 """
1667 updated = self.make_updated_location(self.fileDescriptor.location)
1668 return updated.pathInStore.path
1670 def segregate_parameters(self, parameters: dict[str, Any] | None = None) -> tuple[dict, dict]:
1671 """Segregate the supplied parameters.
1673 This splits the parameters into those understood by the
1674 formatter and those not understood by the formatter.
1676 Any unsupported parameters are assumed to be usable by associated
1677 assemblers.
1679 Parameters
1680 ----------
1681 parameters : `dict`, optional
1682 Parameters with values that have been supplied by the caller
1683 and which might be relevant for the formatter. If `None`
1684 parameters will be read from the registered `FileDescriptor`.
1686 Returns
1687 -------
1688 supported : `dict`
1689 Those parameters supported by this formatter.
1690 unsupported : `dict`
1691 Those parameters not supported by this formatter.
1692 """
1693 if parameters is None: 1693 ↛ 1696line 1693 didn't jump to line 1696 because the condition on line 1693 was always true
1694 parameters = self.fileDescriptor.parameters
1696 if parameters is None:
1697 return {}, {}
1699 if self.unsupportedParameters is None: 1699 ↛ 1704line 1699 didn't jump to line 1704 because the condition on line 1699 was always true
1700 # Support none of the parameters
1701 return {}, parameters.copy()
1703 # Start by assuming all are supported
1704 supported = parameters.copy()
1705 unsupported = {}
1707 # And remove any we know are not supported
1708 for p in set(supported):
1709 if p in self.unsupportedParameters:
1710 unsupported[p] = supported.pop(p)
1712 return supported, unsupported
1714 # Support classic V1 interface.
1715 makeUpdatedLocation = make_updated_location
1716 validateExtension = validate_extension
1717 segregateParameters = segregate_parameters
1718 predictPath = predict_path
1720 # Compatibility with V2 properties.
1721 @property
1722 def write_parameters(self) -> Mapping[str, Any]:
1723 return self.writeParameters
1725 @property
1726 def write_recipes(self) -> Mapping[str, Any]:
1727 return self.writeRecipes
1729 @property
1730 def file_descriptor(self) -> FileDescriptor:
1731 return self.fileDescriptor
1733 @property
1734 def data_id(self) -> DataCoordinate:
1735 return self.dataId
1738class FormatterFactory:
1739 """Factory for `Formatter` instances."""
1741 defaultKey = LookupKey("default")
1742 """Configuration key associated with default write parameter settings."""
1744 writeRecipesKey = LookupKey("write_recipes")
1745 """Configuration key associated with write recipes."""
1747 def __init__(self) -> None:
1748 self._mappingFactory = MappingFactory(Formatter)
1750 def __contains__(self, key: LookupKey | str) -> bool:
1751 """Indicate whether the supplied key is present in the factory.
1753 Parameters
1754 ----------
1755 key : `LookupKey`, `str` or objects with ``name`` attribute
1756 Key to use to lookup in the factory whether a corresponding
1757 formatter is present.
1759 Returns
1760 -------
1761 in : `bool`
1762 `True` if the supplied key is present in the factory.
1763 """
1764 return key in self._mappingFactory
1766 def registerFormatters(self, config: Config, *, universe: DimensionUniverse) -> None:
1767 """Bulk register formatters from a config.
1769 Parameters
1770 ----------
1771 config : `Config`
1772 ``formatters`` section of a configuration.
1773 universe : `DimensionUniverse`, optional
1774 Set of all known dimensions, used to expand and validate any used
1775 in lookup keys.
1777 Notes
1778 -----
1779 The configuration can include one level of hierarchy where an
1780 instrument-specific section can be defined to override more general
1781 template specifications. This is represented in YAML using a
1782 key of form ``instrument<name>`` which can then define templates
1783 that will be returned if a `DatasetRef` contains a matching instrument
1784 name in the data ID.
1786 The config is parsed using the function
1787 `~lsst.daf.butler.configSubset.processLookupConfigs`.
1789 The values for formatter entries can be either a simple string
1790 referring to a python type or a dict representing the formatter and
1791 parameters to be hard-coded into the formatter constructor. For
1792 the dict case the following keys are supported:
1794 - formatter: The python type to be used as the formatter class.
1795 - parameters: A further dict to be passed directly to the
1796 ``write_parameters`` Formatter constructor to seed it.
1797 These parameters are validated at instance creation and not at
1798 configuration.
1800 Additionally, a special ``default`` section can be defined that
1801 uses the formatter type (class) name as the keys and specifies
1802 default write parameters that should be used whenever an instance
1803 of that class is constructed.
1805 .. code-block:: yaml
1807 formatters:
1808 default:
1809 lsst.daf.butler.formatters.example.ExampleFormatter:
1810 max: 10
1811 min: 2
1812 comment: Default comment
1813 calexp: lsst.daf.butler.formatters.example.ExampleFormatter
1814 coadd:
1815 formatter: lsst.daf.butler.formatters.example.ExampleFormatter
1816 parameters:
1817 max: 5
1819 Any time an ``ExampleFormatter`` is constructed it will use those
1820 parameters. If an explicit entry later in the configuration specifies
1821 a different set of parameters, the two will be merged with the later
1822 entry taking priority. In the example above ``calexp`` will use
1823 the default parameters but ``coadd`` will override the value for
1824 ``max``.
1826 Formatter configuration can also include a special section describing
1827 collections of write parameters that can be accessed through a
1828 simple label. This allows common collections of options to be
1829 specified in one place in the configuration and reused later.
1830 The ``write_recipes`` section is indexed by Formatter class name
1831 and each key is the label to associate with the parameters.
1833 .. code-block:: yaml
1835 formatters:
1836 write_recipes:
1837 lsst.obs.base.formatters.fitsExposure.FixExposureFormatter:
1838 lossless:
1839 ...
1840 noCompression:
1841 ...
1843 By convention a formatter that uses write recipes will support a
1844 ``recipe`` write parameter that will refer to a recipe name in
1845 the ``write_recipes`` component. The `Formatter` will be constructed
1846 in the `FormatterFactory` with all the relevant recipes and
1847 will not attempt to filter by looking at ``write_parameters`` in
1848 advance. See the specific formatter documentation for details on
1849 acceptable recipe options.
1850 """
1851 allowed_keys = {"formatter", "parameters"}
1853 contents = processLookupConfigs(config, allow_hierarchy=True, universe=universe)
1855 # Extract any default parameter settings
1856 defaultParameters = contents.get(self.defaultKey, {})
1857 if not isinstance(defaultParameters, Mapping): 1857 ↛ 1858line 1857 didn't jump to line 1858 because the condition on line 1857 was never true
1858 raise RuntimeError(
1859 "Default formatter parameters in config can not be a single string"
1860 f" (got: {type(defaultParameters)})"
1861 )
1863 # Extract any global write recipes -- these are indexed by
1864 # Formatter class name.
1865 writeRecipes = contents.get(self.writeRecipesKey, {})
1866 if isinstance(writeRecipes, str): 1866 ↛ 1867line 1866 didn't jump to line 1867 because the condition on line 1866 was never true
1867 raise RuntimeError(
1868 f"The formatters.{self.writeRecipesKey} section must refer to a dict not '{writeRecipes}'"
1869 )
1871 for key, f in contents.items():
1872 # default is handled in a special way
1873 if key == self.defaultKey:
1874 continue
1875 if key == self.writeRecipesKey:
1876 continue
1878 # Can be a str or a dict.
1879 specificWriteParameters = {}
1880 if isinstance(f, str):
1881 formatter = f
1882 elif isinstance(f, Mapping): 1882 ↛ 1893line 1882 didn't jump to line 1893 because the condition on line 1882 was always true
1883 all_keys = set(f)
1884 unexpected_keys = all_keys - allowed_keys
1885 if unexpected_keys: 1885 ↛ 1886line 1885 didn't jump to line 1886 because the condition on line 1885 was never true
1886 raise ValueError(f"Formatter {key} uses unexpected keys {unexpected_keys} in config")
1887 if "formatter" not in f: 1887 ↛ 1888line 1887 didn't jump to line 1888 because the condition on line 1887 was never true
1888 raise ValueError(f"Mandatory 'formatter' key missing for formatter key {key}")
1889 formatter = f["formatter"]
1890 if "parameters" in f: 1890 ↛ 1896line 1890 didn't jump to line 1896 because the condition on line 1890 was always true
1891 specificWriteParameters = f["parameters"]
1892 else:
1893 raise ValueError(f"Formatter for key {key} has unexpected value: '{f}'")
1895 # Apply any default parameters for this formatter
1896 writeParameters = copy.deepcopy(defaultParameters.get(formatter, {}))
1897 writeParameters.update(specificWriteParameters)
1899 kwargs: dict[str, Any] = {}
1900 if writeParameters:
1901 kwargs["write_parameters"] = writeParameters
1903 if formatter in writeRecipes:
1904 kwargs["write_recipes"] = writeRecipes[formatter]
1906 self.registerFormatter(key, formatter, **kwargs)
1908 def getLookupKeys(self) -> set[LookupKey]:
1909 """Retrieve the look up keys for all the registry entries.
1911 Returns
1912 -------
1913 keys : `set` of `LookupKey`
1914 The keys available for matching in the registry.
1915 """
1916 return self._mappingFactory.getLookupKeys()
1918 def getFormatterClassWithMatch(
1919 self, entity: Entity
1920 ) -> tuple[LookupKey, type[Formatter | FormatterV2], dict[str, Any]]:
1921 """Get the matching formatter class along with the registry key.
1923 Parameters
1924 ----------
1925 entity : `DatasetRef`, `DatasetType`, `StorageClass`, or `str`
1926 Entity to use to determine the formatter to return.
1927 `StorageClass` will be used as a last resort if `DatasetRef`
1928 or `DatasetType` instance is provided. Supports instrument
1929 override if a `DatasetRef` is provided configured with an
1930 ``instrument`` value for the data ID.
1932 Returns
1933 -------
1934 matchKey : `LookupKey`
1935 The key that resulted in the successful match.
1936 formatter : `type`
1937 The class of the registered formatter.
1938 formatter_kwargs : `dict`
1939 Keyword arguments that are associated with this formatter entry.
1940 """
1941 names = (LookupKey(name=entity),) if isinstance(entity, str) else entity._lookupNames()
1942 matchKey, formatter, formatter_kwargs = self._mappingFactory.getClassFromRegistryWithMatch(names)
1943 log.debug(
1944 "Retrieved formatter %s from key '%s' for entity '%s'",
1945 get_full_type_name(formatter),
1946 matchKey,
1947 entity,
1948 )
1950 return matchKey, formatter, formatter_kwargs
1952 def getFormatterClass(self, entity: Entity) -> type:
1953 """Get the matching formatter class.
1955 Parameters
1956 ----------
1957 entity : `DatasetRef`, `DatasetType`, `StorageClass`, or `str`
1958 Entity to use to determine the formatter to return.
1959 `StorageClass` will be used as a last resort if `DatasetRef`
1960 or `DatasetType` instance is provided. Supports instrument
1961 override if a `DatasetRef` is provided configured with an
1962 ``instrument`` value for the data ID.
1964 Returns
1965 -------
1966 formatter : `type`
1967 The class of the registered formatter.
1968 """
1969 _, formatter, _ = self.getFormatterClassWithMatch(entity)
1970 return formatter
1972 def getFormatterWithMatch(
1973 self, entity: Entity, *args: Any, **kwargs: Any
1974 ) -> tuple[LookupKey, Formatter | FormatterV2]:
1975 """Get a new formatter instance along with the matching registry key.
1977 Parameters
1978 ----------
1979 entity : `DatasetRef`, `DatasetType`, `StorageClass`, or `str`
1980 Entity to use to determine the formatter to return.
1981 `StorageClass` will be used as a last resort if `DatasetRef`
1982 or `DatasetType` instance is provided. Supports instrument
1983 override if a `DatasetRef` is provided configured with an
1984 ``instrument`` value for the data ID.
1985 *args : `tuple`
1986 Positional arguments to use pass to the object constructor.
1987 **kwargs
1988 Keyword arguments to pass to object constructor.
1990 Returns
1991 -------
1992 matchKey : `LookupKey`
1993 The key that resulted in the successful match.
1994 formatter : `Formatter`
1995 An instance of the registered formatter.
1996 """
1997 names = (LookupKey(name=entity),) if isinstance(entity, str) else entity._lookupNames()
1998 matchKey, formatter = self._mappingFactory.getFromRegistryWithMatch(names, *args, **kwargs)
1999 log.debug(
2000 "Retrieved formatter %s from key '%s' for entity '%s'",
2001 get_full_type_name(formatter),
2002 matchKey,
2003 entity,
2004 )
2006 return matchKey, formatter
2008 def getFormatter(self, entity: Entity, *args: Any, **kwargs: Any) -> Formatter | FormatterV2:
2009 """Get a new formatter instance.
2011 Parameters
2012 ----------
2013 entity : `DatasetRef`, `DatasetType`, `StorageClass`, or `str`
2014 Entity to use to determine the formatter to return.
2015 `StorageClass` will be used as a last resort if `DatasetRef`
2016 or `DatasetType` instance is provided. Supports instrument
2017 override if a `DatasetRef` is provided configured with an
2018 ``instrument`` value for the data ID.
2019 *args : `tuple`
2020 Positional arguments to use pass to the object constructor.
2021 **kwargs
2022 Keyword arguments to pass to object constructor.
2024 Returns
2025 -------
2026 formatter : `Formatter`
2027 An instance of the registered formatter.
2028 """
2029 _, formatter = self.getFormatterWithMatch(entity, *args, **kwargs)
2030 return formatter
2032 def registerFormatter(
2033 self,
2034 type_: LookupKey | str | StorageClass | DatasetType,
2035 formatter: str,
2036 *,
2037 overwrite: bool = False,
2038 **kwargs: Any,
2039 ) -> None:
2040 """Register a `Formatter`.
2042 Parameters
2043 ----------
2044 type_ : `LookupKey`, `str`, `StorageClass` or `DatasetType`
2045 Type for which this formatter is to be used. If a `LookupKey`
2046 is not provided, one will be constructed from the supplied string
2047 or by using the ``name`` property of the supplied entity.
2048 formatter : `str` or class of type `Formatter`
2049 Identifies a `Formatter` subclass to use for reading and writing
2050 Datasets of this type. Can be a `Formatter` class.
2051 overwrite : `bool`, optional
2052 If `True` an existing entry will be replaced by the new value.
2053 Default is `False`.
2054 **kwargs
2055 Keyword arguments to always pass to object constructor when
2056 retrieved.
2058 Raises
2059 ------
2060 ValueError
2061 Raised if the formatter does not name a valid formatter type and
2062 ``overwrite`` is `False`.
2063 """
2064 self._mappingFactory.placeInRegistry(type_, formatter, overwrite=overwrite, **kwargs)
2067class FormatterV1inV2(FormatterV2):
2068 """An implementation of a V2 formatter that provides a compatibility
2069 interface for V1 formatters.
2071 Parameters
2072 ----------
2073 file_descriptor : `FileDescriptor`, optional
2074 Identifies the file to read or write, and the associated storage
2075 classes and parameter information. Its value can be `None` if the
2076 caller will never call `Formatter.read` or `Formatter.write`.
2077 ref : `DatasetRef`
2078 The dataset associated with this formatter. Should not be a component
2079 dataset ref.
2080 formatter : `Formatter`
2081 A version 1 `Formatter` instance. The V2 formatter layer forwards calls
2082 to this formatter.
2083 write_parameters : `dict`, optional
2084 Any parameters to be hard-coded into this instance to control how
2085 the dataset is serialized.
2086 write_recipes : `dict`, optional
2087 Detailed write Recipes indexed by recipe name.
2088 **kwargs
2089 Additional arguments that will be ignored but allow for
2090 `Formatter` V1 parameters to be given.
2091 """
2093 can_read_from_local_file = True
2094 """This formatter can read from a local file."""
2096 def __init__(
2097 self,
2098 file_descriptor: FileDescriptor,
2099 *,
2100 ref: DatasetRef,
2101 formatter: Formatter,
2102 write_parameters: Mapping[str, Any] | None = None,
2103 write_recipes: Mapping[str, Any] | None = None,
2104 # Compatibility parameters. Unused in v2.
2105 **kwargs: Any,
2106 ):
2107 if not isinstance(formatter, Formatter): 2107 ↛ 2108line 2107 didn't jump to line 2108 because the condition on line 2107 was never true
2108 raise TypeError(f"Formatter parameter was not a V1 formatter (was {type(formatter)})")
2110 # Replace the class property with instance values from this
2111 # V1 formatter so that the V2 __init__ will be able to validate.
2112 self.supported_write_parameters = formatter.supportedWriteParameters # type: ignore
2113 self._formatter = formatter
2115 super().__init__(
2116 file_descriptor, ref=ref, write_parameters=write_parameters, write_recipes=write_recipes
2117 )
2119 def get_write_extension(self) -> str:
2120 ext = self._formatter.extension
2121 return ext if ext is not None else ""
2123 def segregate_parameters(self, parameters: dict[str, Any] | None = None) -> tuple[dict, dict]:
2124 return self._formatter.segregate_parameters(parameters)
2126 def validate_write_recipes( # type: ignore
2127 self,
2128 recipes: Mapping[str, Any] | None,
2129 ) -> Mapping[str, Any] | None:
2130 # This should be a class method but a class method can not work
2131 # for a dynamic shim. Luckily the shim is only used as an instance.
2132 return self._formatter.validate_write_recipes(recipes)
2134 def read_from_local_file(self, path: str, component: str | None = None, expected_size: int = -1) -> Any:
2135 # Need to temporarily override the location since the V1 formatter
2136 # will not know anything about this local file.
2138 # V2 does not have a fromBytes equivalent.
2139 if self._formatter.can_read_bytes(): 2139 ↛ 2140line 2139 didn't jump to line 2140 because the condition on line 2139 was never true
2140 with open(path, "rb") as fd:
2141 serialized_dataset = fd.read()
2142 return self._formatter.fromBytes(serialized_dataset, component=component)
2144 location = Location(None, path)
2145 with self._formatter._updateLocation(location):
2146 try:
2147 result = self._formatter.read(component=component)
2148 except NotImplementedError:
2149 # V1 raises NotImplementedError but V2 is expecting something
2150 # slightly different.
2151 return NotImplemented
2152 return result
2154 def to_bytes(self, in_memory_dataset: Any) -> bytes:
2155 try:
2156 return self._formatter.toBytes(in_memory_dataset)
2157 except NotImplementedError as e:
2158 # V1 raises NotImplementedError but V2 is expecting something
2159 # slightly different.
2160 raise FormatterNotImplementedError(str(e)) from e
2162 def write_local_file(self, in_memory_dataset: Any, uri: ResourcePath) -> None:
2163 with self._formatter._updateLocation(Location(None, uri)):
2164 try:
2165 self._formatter.write(in_memory_dataset)
2166 except NotImplementedError as e:
2167 # V1 raises NotImplementedError but V2 is expecting something
2168 # slightly different.
2169 raise FormatterNotImplementedError(str(e)) from e
2172# Type to use when allowing a Formatter or its class name
2173FormatterParameter: TypeAlias = str | type[Formatter] | Formatter | FormatterV2 | type[FormatterV2]