Coverage for python / lsst / pipe / base / tests / mocks / _storage_class.py: 42%
183 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-30 08:49 +0000
1# This file is part of pipe_base.
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 "ConvertedUnmockedDataset",
32 "MockDataset",
33 "MockDatasetQuantum",
34 "MockStorageClass",
35 "MockStorageClassDelegate",
36 "get_mock_name",
37 "get_original_name",
38 "is_mock_name",
39)
41import sys
42import uuid
43from collections.abc import Callable, Iterable, Mapping
44from typing import Any, cast
46import pydantic
48from lsst.daf.butler import (
49 DataIdValue,
50 DatasetComponent,
51 DatasetRef,
52 DatasetType,
53 Formatter,
54 FormatterFactory,
55 FormatterV2,
56 LookupKey,
57 SerializedDatasetType,
58 StorageClass,
59 StorageClassDelegate,
60 StorageClassFactory,
61)
62from lsst.daf.butler.formatters.json import JsonFormatter
63from lsst.utils.introspection import get_full_type_name
65_NAME_PREFIX: str = "_mock_"
68def get_mock_name(original: str) -> str:
69 """Return the name of the mock storage class, dataset type, or task label
70 for the given original name.
72 Parameters
73 ----------
74 original : `str`
75 Original name.
77 Returns
78 -------
79 name : `str`
80 The name of the mocked version.
81 """
82 return _NAME_PREFIX + original
85def get_original_name(mock: str) -> str:
86 """Return the name of the original storage class, dataset type, or task
87 label that corresponds to the given mock name.
89 Parameters
90 ----------
91 mock : `str`
92 The mocked name.
94 Returns
95 -------
96 original : `str`
97 The original name.
98 """
99 assert mock.startswith(_NAME_PREFIX)
100 return mock.removeprefix(_NAME_PREFIX)
103def is_mock_name(name: str | None) -> bool:
104 """Return whether the given name is that of a mock storage class, dataset
105 type, or task label.
107 Parameters
108 ----------
109 name : `str` or `None`
110 The given name to check.
112 Returns
113 -------
114 is_mock : `bool`
115 Whether the name is for a mock or not.
116 """
117 return name is not None and name.startswith(_NAME_PREFIX)
120# Tests for this module are in the ci_middleware package, where we have easy
121# access to complex real storage classes (and their pytypes) to test against.
124class MockDataset(pydantic.BaseModel):
125 """The in-memory dataset type used by `MockStorageClass`."""
127 dataset_id: uuid.UUID | None
128 """Universal unique identifier for this dataset."""
130 dataset_type: SerializedDatasetType
131 """Butler dataset type or this dataset.
133 See the documentation for ``data_id`` for why this is a
134 `~lsst.daf.butler.SerializedDatasetType` instead of a "real" one.
135 """
137 data_id: dict[str, DataIdValue]
138 """Butler data ID for this dataset.
140 This is a `~lsst.daf.butler.SerializedDataCoordinate` instead of a "real"
141 one for two reasons:
143 - the mock dataset may need to be read from disk in a context in which a
144 `~lsst.daf.butler.DimensionUniverse` is unavailable;
145 - we don't want the complexity of having a separate
146 ``SerializedMockDataCoordinate``.
147 """
149 run: str | None
150 """`~lsst.daf.butler.CollectionType.RUN` collection this dataset belongs
151 to.
152 """
154 quantum: MockDatasetQuantum | None = None
155 """Description of the quantum that produced this dataset.
156 """
158 output_connection_name: str | None = None
159 """The name of the PipelineTask output connection that produced this
160 dataset.
161 """
163 converted_from: MockDataset | None = None
164 """Another `MockDataset` that underwent a storage class conversion to
165 produce this one.
166 """
168 parent: MockDataset | None = None
169 """Another `MockDataset` from which a component was extract to form this
170 one.
171 """
173 parameters: dict[str, str] | None = None
174 """`repr` of all parameters applied when reading this dataset."""
176 int_value: int | None = None
177 """An arbitrary integer value stored in the mock dataset."""
179 str_value: int | None = None
180 """An arbitrary string value stored in the mock dataset."""
182 @property
183 def storage_class(self) -> str:
184 return cast(str, self.dataset_type.storageClass)
186 def make_derived(self, **kwargs: Any) -> MockDataset:
187 """Return a new MockDataset that represents applying some storage class
188 operation to this one.
190 Parameters
191 ----------
192 **kwargs : `~typing.Any`
193 Keyword arguments are fields of `MockDataset` or
194 `~lsst.daf.butler.SerializedDatasetType` to override in the result.
196 Returns
197 -------
198 derived : `MockDataset`
199 The newly-mocked dataset.
200 """
201 dataset_type_updates = {
202 k: kwargs.pop(k) for k in list(kwargs) if k in SerializedDatasetType.model_fields
203 }
204 kwargs.setdefault("dataset_type", self.dataset_type.model_copy(update=dataset_type_updates))
205 # Fields below are those that should not be propagated to the derived
206 # dataset, because they're not about the intrinsic on-disk thing.
207 kwargs.setdefault("converted_from", None)
208 kwargs.setdefault("parent", None)
209 kwargs.setdefault("parameters", None)
210 # Also use setdefault on the ref in case caller wants to override that
211 # directly, but this is expected to be rare enough that it's not worth
212 # it to try to optimize out the work above to make derived_ref.
213 return self.model_copy(update=kwargs)
215 # Work around the fact that Sphinx chokes on Pydantic docstring formatting,
216 # when we inherit those docstrings in our public classes.
217 if "sphinx" in sys.modules:
219 def copy(self, *args: Any, **kwargs: Any) -> Any:
220 """See `pydantic.BaseModel.copy`."""
221 return super().copy(*args, **kwargs)
223 def model_dump(self, *args: Any, **kwargs: Any) -> Any:
224 """See `pydantic.BaseModel.model_dump`."""
225 return super().model_dump(*args, **kwargs)
227 def model_dump_json(self, *args: Any, **kwargs: Any) -> Any:
228 """See `pydantic.BaseModel.model_dump_json`."""
229 return super().model_dump(*args, **kwargs)
231 def model_copy(self, *args: Any, **kwargs: Any) -> Any:
232 """See `pydantic.BaseModel.model_copy`."""
233 return super().model_copy(*args, **kwargs)
235 @classmethod
236 def model_construct(cls, *args: Any, **kwargs: Any) -> Any: # type: ignore[override]
237 """See `pydantic.BaseModel.model_construct`."""
238 return super().model_construct(*args, **kwargs)
240 @classmethod
241 def model_json_schema(cls, *args: Any, **kwargs: Any) -> Any:
242 """See `pydantic.BaseModel.model_json_schema`."""
243 return super().model_json_schema(*args, **kwargs)
245 @classmethod
246 def model_validate(cls, *args: Any, **kwargs: Any) -> Any:
247 """See `pydantic.BaseModel.model_validate`."""
248 return super().model_validate(*args, **kwargs)
250 @classmethod
251 def model_validate_json(cls, *args: Any, **kwargs: Any) -> Any:
252 """See `pydantic.BaseModel.model_validate_json`."""
253 return super().model_validate_json(*args, **kwargs)
255 @classmethod
256 def model_validate_strings(cls, *args: Any, **kwargs: Any) -> Any:
257 """See `pydantic.BaseModel.model_validate_strings`."""
258 return super().model_validate_strings(*args, **kwargs)
261class ConvertedUnmockedDataset(pydantic.BaseModel):
262 """A marker class that represents a conversion from a regular in-memory
263 dataset to a mock storage class.
264 """
266 original_type: str
267 """The full Python type of the original unmocked in-memory dataset."""
269 # Work around the fact that Sphinx chokes on Pydantic docstring formatting,
270 # when we inherit those docstrings in our public classes.
271 if "sphinx" in sys.modules:
273 def copy(self, *args: Any, **kwargs: Any) -> Any:
274 """See `pydantic.BaseModel.copy`."""
275 return super().copy(*args, **kwargs)
277 def model_dump(self, *args: Any, **kwargs: Any) -> Any:
278 """See `pydantic.BaseModel.model_dump`."""
279 return super().model_dump(*args, **kwargs)
281 def model_dump_json(self, *args: Any, **kwargs: Any) -> Any:
282 """See `pydantic.BaseModel.model_dump_json`."""
283 return super().model_dump(*args, **kwargs)
285 def model_copy(self, *args: Any, **kwargs: Any) -> Any:
286 """See `pydantic.BaseModel.model_copy`."""
287 return super().model_copy(*args, **kwargs)
289 @classmethod
290 def model_construct(cls, *args: Any, **kwargs: Any) -> Any: # type: ignore[misc, override]
291 """See `pydantic.BaseModel.model_construct`."""
292 return super().model_construct(*args, **kwargs)
294 @classmethod
295 def model_json_schema(cls, *args: Any, **kwargs: Any) -> Any:
296 """See `pydantic.BaseModel.model_json_schema`."""
297 return super().model_json_schema(*args, **kwargs)
299 @classmethod
300 def model_validate(cls, *args: Any, **kwargs: Any) -> Any:
301 """See `pydantic.BaseModel.model_validate`."""
302 return super().model_validate(*args, **kwargs)
304 @classmethod
305 def model_validate_json(cls, *args: Any, **kwargs: Any) -> Any:
306 """See `pydantic.BaseModel.model_validate_json`."""
307 return super().model_validate_json(*args, **kwargs)
309 @classmethod
310 def model_validate_strings(cls, *args: Any, **kwargs: Any) -> Any:
311 """See `pydantic.BaseModel.model_validate_strings`."""
312 return super().model_validate_strings(*args, **kwargs)
315class MockDatasetQuantum(pydantic.BaseModel):
316 """Description of the quantum that produced a mock dataset.
318 This is also used to represent task-init operations for init-output mock
319 datasets.
320 """
322 task_label: str
323 """Label of the producing PipelineTask in its pipeline."""
325 data_id: dict[str, DataIdValue]
326 """Data ID for the quantum."""
328 inputs: dict[str, list[MockDataset | ConvertedUnmockedDataset]]
329 """Mock datasets provided as input to the quantum.
331 Keys are task-internal connection names, not dataset type names.
332 """
334 # Work around the fact that Sphinx chokes on Pydantic docstring formatting,
335 # when we inherit those docstrings in our public classes.
336 if "sphinx" in sys.modules:
338 def copy(self, *args: Any, **kwargs: Any) -> Any:
339 """See `pydantic.BaseModel.copy`."""
340 return super().copy(*args, **kwargs)
342 def model_dump(self, *args: Any, **kwargs: Any) -> Any:
343 """See `pydantic.BaseModel.model_dump`."""
344 return super().model_dump(*args, **kwargs)
346 def model_dump_json(self, *args: Any, **kwargs: Any) -> Any:
347 """See `pydantic.BaseModel.model_dump_json`."""
348 return super().model_dump(*args, **kwargs)
350 def model_copy(self, *args: Any, **kwargs: Any) -> Any:
351 """See `pydantic.BaseModel.model_copy`."""
352 return super().model_copy(*args, **kwargs)
354 @classmethod
355 def model_construct(cls, *args: Any, **kwargs: Any) -> Any: # type: ignore[misc, override]
356 """See `pydantic.BaseModel.model_construct`."""
357 return super().model_construct(*args, **kwargs)
359 @classmethod
360 def model_json_schema(cls, *args: Any, **kwargs: Any) -> Any:
361 """See `pydantic.BaseModel.model_json_schema`."""
362 return super().model_json_schema(*args, **kwargs)
364 @classmethod
365 def model_validate(cls, *args: Any, **kwargs: Any) -> Any:
366 """See `pydantic.BaseModel.model_validate`."""
367 return super().model_validate(*args, **kwargs)
369 @classmethod
370 def model_validate_json(cls, *args: Any, **kwargs: Any) -> Any:
371 """See `pydantic.BaseModel.model_validate_json`."""
372 return super().model_validate_json(*args, **kwargs)
374 @classmethod
375 def model_validate_strings(cls, *args: Any, **kwargs: Any) -> Any:
376 """See `pydantic.BaseModel.model_validate_strings`."""
377 return super().model_validate_strings(*args, **kwargs)
380MockDataset.model_rebuild()
383class MockStorageClassDelegate(StorageClassDelegate):
384 """Implementation of the `~lsst.daf.butler.StorageClassDelegate` interface
385 for mock datasets.
387 This class does not implement assembly and disassembly just because it's
388 not needed right now. That could be added in the future with some
389 additional tracking attributes in `MockDataset`.
390 """
392 def assemble(self, components: dict[str, Any], pytype: type | None = None) -> MockDataset:
393 # Docstring inherited.
394 raise NotImplementedError("Mock storage classes do not implement assembly.")
396 def getComponent(self, composite: Any, componentName: str) -> Any:
397 # Docstring inherited.
398 assert isinstance(composite, MockDataset), (
399 f"MockStorageClassDelegate given a non-mock dataset {composite!r}."
400 )
401 return composite.make_derived(
402 name=f"{composite.dataset_type.name}.{componentName}",
403 storageClass=self.storageClass.allComponents()[componentName].name,
404 parentStorageClass=self.storageClass.name,
405 parent=composite,
406 )
408 def disassemble(
409 self, composite: Any, subset: Iterable | None = None, override: Any | None = None
410 ) -> dict[str, DatasetComponent]:
411 # Docstring inherited.
412 raise NotImplementedError("Mock storage classes do not implement disassembly.")
414 def handleParameters(self, inMemoryDataset: Any, parameters: Mapping[str, Any] | None = None) -> Any:
415 # Docstring inherited.
416 assert isinstance(inMemoryDataset, MockDataset), (
417 f"MockStorageClassDelegate given a non-mock dataset {inMemoryDataset!r}."
418 )
419 if not parameters:
420 return inMemoryDataset
421 return inMemoryDataset.make_derived(parameters={k: repr(v) for k, v in parameters.items()})
424class MockStorageClass(StorageClass):
425 """A reimplementation of `lsst.daf.butler.StorageClass` for mock datasets.
427 Parameters
428 ----------
429 original : `~lsst.daf.butler.StorageClass`
430 The original storage class.
431 factory : `~lsst.daf.butler.StorageClassFactory` or `None`, optional
432 Storage class factory to use. If `None` the default factory is used.
434 Notes
435 -----
436 Each `MockStorageClass` instance corresponds to a real "original" storage
437 class, with components and conversions that are mocks of the original's
438 components and conversions. The ``pytype`` for all `MockStorageClass`
439 instances is `MockDataset`.
440 """
442 def __init__(self, original: StorageClass, factory: StorageClassFactory | None = None):
443 name = get_mock_name(original.name)
444 if factory is None:
445 factory = StorageClassFactory()
446 super().__init__(
447 name=name,
448 pytype=MockDataset,
449 components={
450 k: self.get_or_register_mock(v.name, factory) for k, v in original.components.items()
451 },
452 derivedComponents={
453 k: self.get_or_register_mock(v.name, factory) for k, v in original.derivedComponents.items()
454 },
455 parameters=frozenset(original.parameters),
456 delegate=get_full_type_name(MockStorageClassDelegate),
457 # Conversions work differently for mock storage classes, since they
458 # all have the same pytype: we use the original storage class being
459 # mocked to see if we can convert, then just make a new MockDataset
460 # that points back to the original.
461 converters={},
462 )
463 self.original = original
464 # Make certain no one tries to use the converters.
465 self._converters = None # type: ignore
467 def _get_converters_by_type(self) -> dict[type, Callable[[Any], Any]]:
468 # Docstring inherited.
469 raise NotImplementedError("MockStorageClass does not use converters.")
471 @classmethod
472 def get_or_register_mock(
473 cls, original: str, factory: StorageClassFactory | None = None
474 ) -> MockStorageClass:
475 """Return a mock storage class for the given original storage class,
476 creating and registering it if necessary.
478 Parameters
479 ----------
480 original : `str`
481 Name of the original storage class to be mocked.
482 factory : `~lsst.daf.butler.StorageClassFactory`, optional
483 Storage class factory singleton instance.
485 Returns
486 -------
487 mock : `MockStorageClass`
488 New storage class that mocks ``original``.
489 """
490 name = get_mock_name(original)
491 if factory is None:
492 factory = StorageClassFactory()
493 if name in factory:
494 return cast(MockStorageClass, factory.getStorageClass(name))
495 else:
496 result = cls(factory.getStorageClass(original), factory)
497 factory.registerStorageClass(result)
498 return result
500 def allComponents(self) -> Mapping[str, MockStorageClass]:
501 # Docstring inherited.
502 return cast(Mapping[str, MockStorageClass], super().allComponents())
504 @property
505 def components(self) -> Mapping[str, MockStorageClass]:
506 # Docstring inherited.
507 return cast(Mapping[str, MockStorageClass], super().components)
509 @property
510 def derivedComponents(self) -> Mapping[str, MockStorageClass]:
511 # Docstring inherited.
512 return cast(Mapping[str, MockStorageClass], super().derivedComponents)
514 def can_convert(self, other: StorageClass) -> bool:
515 # Docstring inherited.
516 if not isinstance(other, MockStorageClass):
517 # Allow conversions from an original type (and others compatible
518 # with it) to a mock, to allow for cases where an upstream task
519 # did not use a mock to write something but the downstream one is
520 # trying to us a mock to read it.
521 return self.original.can_convert(other)
522 return self.original.can_convert(other.original)
524 def coerce_type(self, incorrect: Any) -> Any:
525 # Docstring inherited.
526 if not isinstance(incorrect, MockDataset):
527 if isinstance(incorrect, ConvertedUnmockedDataset):
528 return incorrect
529 return ConvertedUnmockedDataset(original_type=get_full_type_name(incorrect))
530 factory = StorageClassFactory()
531 other_storage_class = factory.getStorageClass(incorrect.storage_class)
532 assert isinstance(other_storage_class, MockStorageClass), "Should not get a MockDataset otherwise."
533 if other_storage_class.name == self.name:
534 return incorrect
535 if not self.can_convert(other_storage_class):
536 raise TypeError(
537 f"Mocked storage class {self.original.name!r} cannot convert from "
538 f"{other_storage_class.original.name!r}."
539 )
540 return incorrect.make_derived(storageClass=self.name, converted_from=incorrect)
542 @staticmethod
543 def mock_dataset_type(original_type: DatasetType) -> DatasetType:
544 """Replace a dataset type with a version that uses a mock storage class
545 and name.
547 Parameters
548 ----------
549 original_type : `lsst.daf.butler.DatasetType`
550 Original dataset type to be mocked.
552 Returns
553 -------
554 mock_type : `lsst.daf.butler.DatasetType`
555 A mock version of the dataset type, with name and storage class
556 changed and everything else unchanged.
557 """
558 mock_storage_class = MockStorageClass.get_or_register_mock(original_type.storageClass_name)
559 mock_parent_storage_class = None
560 if original_type.parentStorageClass is not None:
561 mock_parent_storage_class = MockStorageClass.get_or_register_mock(
562 original_type.parentStorageClass.name
563 )
564 return DatasetType(
565 get_mock_name(original_type.name),
566 original_type.dimensions,
567 mock_storage_class,
568 isCalibration=original_type.isCalibration(),
569 parentStorageClass=mock_parent_storage_class,
570 )
572 @staticmethod
573 def mock_dataset_refs(original_refs: Iterable[DatasetRef]) -> list[DatasetRef]:
574 """Replace dataset references with versions that uses a mock storage
575 class and dataset type name.
577 Parameters
578 ----------
579 original_refs : `~collections.abc.Iterable` [ \
580 `lsst.daf.butler.DatasetRef` ]
581 Original dataset references to be mocked.
583 Returns
584 -------
585 mock_refs : `list` [ `lsst.daf.butler.DatasetRef` ]
586 Mocked version of the dataset references, with dataset type name
587 and storage class changed and everything else unchanged.
588 """
589 original_refs = list(original_refs)
590 if not original_refs:
591 return original_refs
592 dataset_type = MockStorageClass.mock_dataset_type(original_refs[0].datasetType)
593 return [
594 DatasetRef(dataset_type, original_ref.dataId, run=original_ref.run, id=original_ref.id)
595 for original_ref in original_refs
596 ]
598 @staticmethod
599 def unmock_dataset_type(mock_type: DatasetType) -> DatasetType:
600 """Replace a mock dataset type with the original one it was created
601 from.
603 Parameters
604 ----------
605 mock_type : `lsst.daf.butler.DatasetType`
606 A dataset type with a mocked name and storage class.
608 Returns
609 -------
610 original_type : `lsst.daf.butler.DatasetType`
611 The original dataset type.
612 """
613 storage_class = mock_type.storageClass
614 parent_storage_class = mock_type.parentStorageClass
615 if isinstance(storage_class, MockStorageClass):
616 storage_class = storage_class.original
617 if parent_storage_class is not None and isinstance(parent_storage_class, MockStorageClass):
618 parent_storage_class = parent_storage_class.original
619 return DatasetType(
620 get_original_name(mock_type.name),
621 mock_type.dimensions,
622 storage_class,
623 isCalibration=mock_type.isCalibration(),
624 parentStorageClass=parent_storage_class,
625 )
627 @staticmethod
628 def unmock_dataset_refs(mock_refs: Iterable[DatasetRef]) -> list[DatasetRef]:
629 """Replace dataset references with versions that do not use a mock
630 storage class and dataset type name.
632 Parameters
633 ----------
634 mock_refs : `~collections.abc.Iterable` [ \
635 `lsst.daf.butler.DatasetRef` ]
636 Dataset references that use a mocked dataset type name and storage
637 class.
639 Returns
640 -------
641 original_refs : `list` [ `lsst.daf.butler.DatasetRef` ]
642 The original dataset references.
643 """
644 mock_refs = list(mock_refs)
645 if not mock_refs:
646 return mock_refs
647 dataset_type = MockStorageClass.unmock_dataset_type(mock_refs[0].datasetType)
648 return [
649 DatasetRef(dataset_type, mock_ref.dataId, run=mock_ref.run, id=mock_ref.id)
650 for mock_ref in mock_refs
651 ]
654def _monkeypatch_daf_butler() -> None:
655 """Replace methods in daf_butler's StorageClassFactory and FormatterFactory
656 classes to automatically recognize mock storage classes.
658 This monkey-patching is executed when the `lsst.pipe.base.tests.mocks`
659 package is imported, and it affects all butler instances created before or
660 after that imported.
661 """
662 original_get_storage_class = StorageClassFactory.getStorageClass
664 def new_get_storage_class(self: StorageClassFactory, storageClassName: str) -> StorageClass:
665 try:
666 return original_get_storage_class(self, storageClassName)
667 except KeyError:
668 if is_mock_name(storageClassName):
669 return MockStorageClass.get_or_register_mock(get_original_name(storageClassName))
670 raise
672 StorageClassFactory.getStorageClass = new_get_storage_class # type: ignore
674 del new_get_storage_class
676 original_get_formatter_class_with_match = FormatterFactory.getFormatterClassWithMatch
678 def new_get_formatter_class_with_match(
679 self: FormatterFactory, entity: Any
680 ) -> tuple[LookupKey, type[Formatter | FormatterV2], dict[str, Any]]:
681 try:
682 return original_get_formatter_class_with_match(self, entity)
683 except KeyError:
684 lookup_keys = (LookupKey(name=entity),) if isinstance(entity, str) else entity._lookupNames()
685 for key in lookup_keys:
686 # This matches mock dataset type names before mock storage
687 # classes, and it would even match some regular dataset types
688 # that are automatic connections (logs, configs, metadata) of
689 # mocked tasks. The latter would be a problem, except that
690 # those should have already matched in the try block above.
691 if is_mock_name(key.name):
692 return (key, JsonFormatter, {})
693 raise
695 FormatterFactory.getFormatterClassWithMatch = new_get_formatter_class_with_match # type: ignore
697 del new_get_formatter_class_with_match
699 original_get_formatter_with_match = FormatterFactory.getFormatterWithMatch
701 def new_get_formatter_with_match(
702 self: FormatterFactory, entity: Any, *args: Any, **kwargs: Any
703 ) -> tuple[LookupKey, Formatter | FormatterV2]:
704 try:
705 return original_get_formatter_with_match(self, entity, *args, **kwargs)
706 except KeyError:
707 lookup_keys = (LookupKey(name=entity),) if isinstance(entity, str) else entity._lookupNames()
708 for key in lookup_keys:
709 if is_mock_name(key.name):
710 return (key, JsonFormatter(*args, **kwargs))
711 raise
713 FormatterFactory.getFormatterWithMatch = new_get_formatter_with_match # type: ignore
715 del new_get_formatter_with_match
718_monkeypatch_daf_butler()