Coverage for python / lsst / pipe / base / single_quantum_executor.py: 11%
220 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-18 01:04 -0800
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/>.
28__all__ = ["SingleQuantumExecutor"]
30import logging
31import time
32import uuid
33from collections import defaultdict
34from collections.abc import Callable, Mapping
35from itertools import chain
36from typing import Any
38from lsst.daf.butler import (
39 Butler,
40 DatasetRef,
41 DatasetType,
42 LimitedButler,
43 NamedKeyDict,
44 Quantum,
45)
46from lsst.daf.butler.logging import ButlerLogRecords
47from lsst.utils.introspection import get_full_type_name
48from lsst.utils.timer import logInfo
50from ._quantumContext import ExecutionResources, QuantumContext
51from ._status import (
52 AnnotatedPartialOutputsError,
53 ExceptionInfo,
54 InvalidQuantumError,
55 NoWorkFound,
56 QuantumSuccessCaveats,
57)
58from .connections import AdjustQuantumHelper
59from .log_capture import LogCapture, _ExecutionLogRecordsExtra
60from .pipeline_graph import TaskNode
61from .pipelineTask import PipelineTask
62from .quantum_graph_executor import QuantumExecutionResult, QuantumExecutor
63from .quantum_reports import QuantumReport
64from .task import _TASK_FULL_METADATA_TYPE, _TASK_METADATA_TYPE
65from .taskFactory import TaskFactory
67_LOG = logging.getLogger(__name__)
70class SingleQuantumExecutor(QuantumExecutor):
71 """Executor class which runs one Quantum at a time.
73 Parameters
74 ----------
75 butler : `~lsst.daf.butler.Butler` or `None`, optional
76 Data butler, `None` means that a limited butler should be used instead.
77 task_factory : `.TaskFactory`, optional
78 Instance of a task factory. Defaults to a new instance of
79 `lsst.pipe.base.TaskFactory`.
80 skip_existing_in : `str` or `~collections.abc.Iterable` [ `str` ]
81 Expressions representing the collections to search for existing output
82 datasets. See :ref:`daf_butler_ordered_collection_searches` for allowed
83 types. This class only checks for the presence of butler output run in
84 the list of collections. If the output run is present in the list then
85 the quanta whose complete outputs exist in the output run will be
86 skipped. `None` or empty string/sequence disables skipping.
87 clobber_outputs : `bool`, optional
88 If `True`, then outputs from a quantum that exist in output run
89 collection will be removed prior to executing a quantum. If
90 ``skip_existing_in`` contains output run, then only partial outputs
91 from a quantum will be removed. Only used when ``butler`` is not
92 `None`.
93 enable_lsst_debug : `bool`, optional
94 Enable debugging with ``lsstDebug`` facility for a task.
95 limited_butler_factory : `Callable`, optional
96 A method that creates a `~lsst.daf.butler.LimitedButler` instance for a
97 given Quantum. This parameter must be defined if ``butler`` is `None`.
98 If ``butler`` is not `None` then this parameter is ignored.
99 resources : `.ExecutionResources`, optional
100 The resources available to this quantum when executing.
101 skip_existing : `bool`, optional
102 If `True`, skip quanta whose metadata datasets are already stored.
103 Unlike ``skip_existing_in``, this works with limited butlers as well as
104 full butlers. Always set to `True` if ``skip_existing_in`` matches
105 ``butler.run``.
106 assume_no_existing_outputs : `bool`, optional
107 If `True`, assume preexisting outputs are impossible (e.g. because this
108 is known by higher-level code to be a new ``RUN`` collection), and do
109 not look for them. This causes the ``skip_existing`` and
110 ``clobber_outputs`` options to be ignored, but unlike just setting both
111 of those to `False`, it also avoids all dataset existence checks.
112 raise_on_partial_outputs : `bool`, optional
113 If `True` raise exceptions chained by `.AnnotatedPartialOutputsError`
114 immediately, instead of considering the partial result a success and
115 continuing to run downstream tasks.
116 job_metadata : `~collections.abc.Mapping`
117 Mapping with extra metadata to embed within the quantum metadata under
118 the "job" key. This is intended to correspond to information common
119 to all quanta being executed in a single process, such as the time
120 taken to load the quantum graph in a BPS job.
121 """
123 def __init__(
124 self,
125 *,
126 butler: Butler | None = None,
127 task_factory: TaskFactory | None = None,
128 skip_existing_in: Any = None,
129 clobber_outputs: bool = False,
130 enable_lsst_debug: bool = False,
131 limited_butler_factory: Callable[[Quantum], LimitedButler] | None = None,
132 resources: ExecutionResources | None = None,
133 skip_existing: bool = False,
134 assume_no_existing_outputs: bool = False,
135 raise_on_partial_outputs: bool = True,
136 job_metadata: Mapping[str, int | str | float] | None = None,
137 ):
138 self._butler = butler
139 self._task_factory = task_factory if task_factory is not None else TaskFactory()
140 self._clobber_outputs = clobber_outputs
141 self._enable_lsst_debug = enable_lsst_debug
142 self._limited_butler_factory = limited_butler_factory
143 self._resources = resources
144 self._assume_no_existing_outputs = assume_no_existing_outputs
145 self._raise_on_partial_outputs = raise_on_partial_outputs
146 self._job_metadata = job_metadata
148 if self._butler is None:
149 assert limited_butler_factory is not None, "limited_butler_factory is needed when butler is None"
151 # Find whether output run is in skip_existing_in.
152 self._skip_existing = skip_existing
153 if self._butler is not None and skip_existing_in and not self._skip_existing:
154 self._skip_existing = self._butler.run in self._butler.collections.query(
155 skip_existing_in, flatten_chains=True
156 )
157 self._previous_process_quanta: list[uuid.UUID] = []
159 def execute(
160 self,
161 task_node: TaskNode,
162 /,
163 quantum: Quantum,
164 quantum_id: uuid.UUID | None = None,
165 *,
166 log_records: ButlerLogRecords | None = None,
167 ) -> QuantumExecutionResult:
168 # Docstring inherited from QuantumExecutor.execute
169 if self._butler is not None:
170 self._butler.registry.refresh()
171 return self._execute(task_node, quantum, quantum_id=quantum_id, log_records=log_records)
173 def _execute(
174 self,
175 task_node: TaskNode,
176 /,
177 quantum: Quantum,
178 quantum_id: uuid.UUID | None = None,
179 *,
180 log_records: ButlerLogRecords | None = None,
181 ) -> QuantumExecutionResult:
182 """Execute the quantum.
184 Internal implementation of `execute()`.
185 """
186 # Make a limited butler instance if needed.
187 limited_butler: LimitedButler
188 used_butler_factory = False
189 if self._butler is not None:
190 limited_butler = self._butler
191 else:
192 # We check this in constructor, but mypy needs this check here.
193 assert self._limited_butler_factory is not None
194 limited_butler = self._limited_butler_factory(quantum)
195 used_butler_factory = True
197 try:
198 return self._execute_with_limited_butler(
199 task_node, limited_butler, quantum=quantum, quantum_id=quantum_id, log_records=log_records
200 )
201 finally:
202 if used_butler_factory:
203 limited_butler.close()
205 def _execute_with_limited_butler(
206 self,
207 task_node: TaskNode,
208 limited_butler: LimitedButler,
209 /,
210 quantum: Quantum,
211 quantum_id: uuid.UUID | None = None,
212 *,
213 log_records: ButlerLogRecords | None = None,
214 ) -> QuantumExecutionResult:
215 startTime = time.time()
216 assert quantum.dataId is not None, "Quantum DataId cannot be None"
217 report = QuantumReport(quantumId=quantum_id, dataId=quantum.dataId, taskLabel=task_node.label)
218 if self._butler is not None:
219 log_capture = LogCapture.from_full(self._butler)
220 else:
221 log_capture = LogCapture.from_limited(limited_butler)
222 with log_capture.capture_logging(task_node, quantum, records=log_records) as captureLog:
223 # Save detailed resource usage before task start to metadata.
224 quantumMetadata = _TASK_METADATA_TYPE()
225 logInfo(None, "prep", metadata=quantumMetadata) # type: ignore[arg-type]
227 _LOG.info(
228 "Preparing execution of quantum for label=%s dataId=%s.", task_node.label, quantum.dataId
229 )
231 # check whether to skip or delete old outputs, if it returns True
232 # or raises an exception do not try to store logs, as they may be
233 # already in butler.
234 captureLog.store = False
235 if self._check_existing_outputs(quantum, task_node, limited_butler, captureLog.extra):
236 _LOG.info(
237 "Skipping already-successful quantum for label=%s dataId=%s.",
238 task_node.label,
239 quantum.dataId,
240 )
241 return QuantumExecutionResult(quantum, report, skipped_existing=True, adjusted_no_work=False)
242 captureLog.store = True
244 captureLog.extra.previous_process_quanta.extend(self._previous_process_quanta)
245 if quantum_id is not None:
246 self._previous_process_quanta.append(quantum_id)
247 try:
248 quantum = self._updated_quantum_inputs(quantum, task_node, limited_butler)
249 except NoWorkFound as exc:
250 _LOG.info(
251 "Nothing to do for task '%s' on quantum %s; saving metadata and skipping: %s",
252 task_node.label,
253 quantum.dataId,
254 str(exc),
255 )
256 quantumMetadata["caveats"] = QuantumSuccessCaveats.from_adjust_quantum_no_work().value
257 # Make empty metadata that looks something like what a
258 # do-nothing task would write (but we don't bother with empty
259 # nested PropertySets for subtasks). This is slightly
260 # duplicative with logic in pipe_base that we can't easily call
261 # from here; we'll fix this on DM-29761.
262 logInfo(None, "end", metadata=quantumMetadata) # type: ignore[arg-type]
263 fullMetadata = _TASK_FULL_METADATA_TYPE()
264 fullMetadata[task_node.label] = _TASK_METADATA_TYPE()
265 fullMetadata["quantum"] = quantumMetadata
266 if self._job_metadata is not None:
267 fullMetadata["job"] = self._job_metadata
268 self._write_metadata(quantum, fullMetadata, task_node, limited_butler)
269 return QuantumExecutionResult(quantum, report, skipped_existing=False, adjusted_no_work=True)
271 # enable lsstDebug debugging
272 if self._enable_lsst_debug:
273 try:
274 _LOG.debug("Will try to import debug.py")
275 import debug # type: ignore # noqa:F401
276 except ImportError:
277 _LOG.warning("No 'debug' module found.")
279 # Ensure that we are executing a frozen config
280 task_node.config.freeze()
281 logInfo(None, "init", metadata=quantumMetadata) # type: ignore[arg-type]
282 init_input_refs = list(quantum.initInputs.values())
284 _LOG.info(
285 "Constructing task and executing quantum for label=%s dataId=%s.",
286 task_node.label,
287 quantum.dataId,
288 )
289 task = self._task_factory.makeTask(task_node, limited_butler, init_input_refs)
290 logInfo(None, "start", metadata=quantumMetadata) # type: ignore[arg-type]
291 outputs_put: list[uuid.UUID] = []
292 try:
293 with limited_butler.record_metrics() as butler_metrics:
294 caveats = self._run_quantum(
295 task, quantum, task_node, limited_butler, quantum_id=quantum_id, ids_put=outputs_put
296 )
297 except Exception as e:
298 _LOG.error(
299 "Execution of task '%s' on quantum %s failed. Exception %s: %s",
300 task_node.label,
301 quantum.dataId,
302 e.__class__.__name__,
303 str(e),
304 )
305 captureLog.extra.exception = ExceptionInfo(
306 type_name=get_full_type_name(e),
307 message=str(e),
308 metadata={},
309 )
310 raise
311 else:
312 quantumMetadata["butler_metrics"] = butler_metrics.model_dump()
313 quantumMetadata["caveats"] = caveats.value
314 # Stringify the UUID for easier compatibility with
315 # PropertyList.
316 finally:
317 logInfo(None, "end", metadata=quantumMetadata) # type: ignore[arg-type]
318 fullMetadata = task.getFullMetadata()
319 quantumMetadata["outputs"] = [str(output) for output in outputs_put]
320 fullMetadata["quantum"] = quantumMetadata
321 if self._job_metadata is not None:
322 fullMetadata["job"] = self._job_metadata
323 captureLog.extra.metadata = fullMetadata
324 self._write_metadata(quantum, fullMetadata, task_node, limited_butler)
325 stopTime = time.time()
326 _LOG.info(
327 "Execution of task '%s' on quantum %s took %.3f seconds",
328 task_node.label,
329 quantum.dataId,
330 stopTime - startTime,
331 )
332 return QuantumExecutionResult(
333 quantum,
334 report,
335 skipped_existing=False,
336 adjusted_no_work=False,
337 task_metadata=fullMetadata,
338 )
340 def _check_existing_outputs(
341 self,
342 quantum: Quantum,
343 task_node: TaskNode,
344 /,
345 limited_butler: LimitedButler,
346 log_extra: _ExecutionLogRecordsExtra,
347 ) -> bool:
348 """Decide whether this quantum needs to be executed.
350 If only partial outputs exist then they are removed if
351 ``clobberOutputs`` is True, otherwise an exception is raised.
353 The ``LimitedButler`` is used for everything, and should be set to
354 ``self.butler`` if no separate ``LimitedButler`` is available.
356 Parameters
357 ----------
358 quantum : `~lsst.daf.butler.Quantum`
359 Quantum to check for existing outputs.
360 task_node : `~.pipeline_graph.TaskNode`
361 Task definition structure.
362 limited_butler : `~lsst.daf.butler.LimitedButler`
363 Butler to use for querying and clobbering.
364 log_extra : `.log_capture.TaskLogRecordsExtra`
365 Extra information to attach to log records.
367 Returns
368 -------
369 exist : `bool`
370 `True` if ``self.skipExisting`` is defined, and a previous
371 execution of this quanta appears to have completed successfully
372 (either because metadata was written or all datasets were written).
373 `False` otherwise.
375 Raises
376 ------
377 RuntimeError
378 Raised if some outputs exist and some not.
379 """
380 if self._assume_no_existing_outputs:
381 return False
383 if self._skip_existing:
384 _LOG.debug(
385 "Checking existence of metadata from previous execution of label=%s dataId=%s.",
386 task_node.label,
387 quantum.dataId,
388 )
389 # Metadata output exists; this is sufficient to assume the previous
390 # run was successful and should be skipped.
391 [metadata_ref] = quantum.outputs[task_node.metadata_output.dataset_type_name]
392 if metadata_ref is not None:
393 if limited_butler.stored(metadata_ref):
394 return True
396 # Find and prune (partial) outputs if `self.clobberOutputs` is set.
397 _LOG.debug(
398 "Looking for existing outputs in the way for label=%s dataId=%s.", task_node.label, quantum.dataId
399 )
400 ref_dict = limited_butler.stored_many(chain.from_iterable(quantum.outputs.values()))
401 if task_node.log_output is not None:
402 (log_ref,) = quantum.outputs[task_node.log_output.dataset_type_name]
403 if ref_dict[log_ref]:
404 _LOG.debug(
405 "Attaching logs from previous attempt on label=%s dataId=%s.",
406 task_node.label,
407 quantum.dataId,
408 )
409 log_extra.attach_previous_attempt(limited_butler.get(log_ref))
410 existingRefs = [ref for ref, exists in ref_dict.items() if exists]
411 missingRefs = [ref for ref, exists in ref_dict.items() if not exists]
412 if existingRefs:
413 if not missingRefs:
414 # Full outputs exist.
415 if self._skip_existing:
416 return True
417 elif self._clobber_outputs:
418 _LOG.info("Removing complete outputs for quantum %s: %s", quantum, existingRefs)
419 limited_butler.pruneDatasets(existingRefs, disassociate=True, unstore=True, purge=True)
420 else:
421 raise RuntimeError(
422 f"Complete outputs exists for a quantum {quantum} "
423 "and neither clobberOutputs nor skipExisting is set: "
424 f"existingRefs={existingRefs}"
425 )
426 else:
427 # Partial outputs from a failed quantum.
428 _LOG.debug(
429 "Partial outputs exist for quantum %s existingRefs=%s missingRefs=%s",
430 quantum,
431 existingRefs,
432 missingRefs,
433 )
434 if self._clobber_outputs:
435 # only prune
436 _LOG.info("Removing partial outputs for task %s: %s", task_node.label, existingRefs)
437 limited_butler.pruneDatasets(existingRefs, disassociate=True, unstore=True, purge=True)
438 return False
439 else:
440 raise RuntimeError(
441 "Registry inconsistency while checking for existing quantum outputs:"
442 f" quantum={quantum} existingRefs={existingRefs}"
443 f" missingRefs={missingRefs}"
444 )
446 # By default always execute.
447 return False
449 def _updated_quantum_inputs(
450 self, quantum: Quantum, task_node: TaskNode, /, limited_butler: LimitedButler
451 ) -> Quantum:
452 """Update quantum with extra information, returns a new updated
453 Quantum.
455 Some methods may require input DatasetRefs to have non-None
456 ``dataset_id``, but in case of intermediate dataset it may not be
457 filled during QuantumGraph construction. This method will retrieve
458 missing info from registry.
460 Parameters
461 ----------
462 quantum : `~lsst.daf.butler.Quantum`
463 Single Quantum instance.
464 task_node : `~.pipeline_graph.TaskNode`
465 Task definition structure.
466 limited_butler : `~lsst.daf.butler.LimitedButler`
467 Butler to use for querying.
469 Returns
470 -------
471 update : `~lsst.daf.butler.Quantum`
472 Updated Quantum instance.
473 """
474 anyChanges = False
475 updatedInputs: defaultdict[DatasetType, list] = defaultdict(list)
476 for key, refsForDatasetType in quantum.inputs.items():
477 _LOG.debug(
478 "Checking existence of input '%s' for label=%s dataId=%s.",
479 key.name,
480 task_node.label,
481 quantum.dataId,
482 )
483 toCheck = []
484 newRefsForDatasetType = updatedInputs[key]
485 for ref in refsForDatasetType:
486 if self._should_assume_exists(quantum, ref):
487 newRefsForDatasetType.append(ref)
488 else:
489 toCheck.append(ref)
490 if not toCheck:
491 _LOG.debug(
492 "Assuming overall input '%s' is present without checks for label=%s dataId=%s.",
493 key.name,
494 task_node.label,
495 quantum.dataId,
496 )
497 continue
498 stored = limited_butler.stored_many(toCheck)
499 for ref in toCheck:
500 if stored[ref]:
501 newRefsForDatasetType.append(ref)
502 else:
503 # This should only happen if a predicted intermediate was
504 # not actually produced upstream, but
505 # datastore misconfigurations can unfortunately also land
506 # us here.
507 _LOG.info("No dataset artifact found for %s", ref)
508 continue
509 if len(newRefsForDatasetType) != len(refsForDatasetType):
510 anyChanges = True
511 # If we removed any input datasets, let the task check if it has enough
512 # to proceed and/or prune related datasets that it also doesn't
513 # need/produce anymore. It will raise NoWorkFound if it can't run,
514 # which we'll let propagate up. This is exactly what we run during QG
515 # generation, because a task shouldn't care whether an input is missing
516 # because some previous task didn't produce it, or because it just
517 # wasn't there during QG generation.
518 namedUpdatedInputs = NamedKeyDict[DatasetType, list[DatasetRef]](updatedInputs.items())
519 helper = AdjustQuantumHelper(namedUpdatedInputs, quantum.outputs)
520 if anyChanges:
521 _LOG.debug("Running adjustQuantum for label=%s dataId=%s.", task_node.label, quantum.dataId)
522 assert quantum.dataId is not None, "Quantum DataId cannot be None"
523 helper.adjust_in_place(task_node.get_connections(), label=task_node.label, data_id=quantum.dataId)
524 return Quantum(
525 taskName=quantum.taskName,
526 taskClass=quantum.taskClass,
527 dataId=quantum.dataId,
528 initInputs=quantum.initInputs,
529 inputs=helper.inputs,
530 outputs=helper.outputs,
531 )
533 def _run_quantum(
534 self,
535 task: PipelineTask,
536 quantum: Quantum,
537 task_node: TaskNode,
538 /,
539 limited_butler: LimitedButler,
540 quantum_id: uuid.UUID | None,
541 ids_put: list[uuid.UUID],
542 ) -> QuantumSuccessCaveats:
543 """Execute task on a single quantum.
545 Parameters
546 ----------
547 task : `PipelineTask`
548 Task object.
549 quantum : `~lsst.daf.butler.Quantum`
550 Single Quantum instance.
551 task_node : `~.pipeline_graph.TaskNode`
552 Task definition structure.
553 limited_butler : `~lsst.daf.butler.LimitedButler`
554 Butler to use for dataset I/O.
555 quantum_id : `uuid.UUID` or `None`
556 ID of the quantum being executed.
557 ids_put : list[ `uuid.UUID` ]
558 List to be populated with the dataset IDs that were written by this
559 quantum. This is an output parameter in order to allow it to be
560 populated even when an exception is raised.
562 Returns
563 -------
564 flags : `QuantumSuccessCaveats`
565 Flags that describe qualified successes.
566 """
567 flags = QuantumSuccessCaveats.NO_CAVEATS
569 # Create a butler that operates in the context of a quantum
570 butlerQC = QuantumContext(limited_butler, quantum, resources=self._resources, quantum_id=quantum_id)
572 # Get the input and output references for the task
573 inputRefs, outputRefs = task_node.get_connections().buildDatasetRefs(quantum)
575 # Call task runQuantum() method.
576 try:
577 task.runQuantum(butlerQC, inputRefs, outputRefs)
578 except NoWorkFound as err:
579 # Not an error, just an early exit.
580 _LOG.info(
581 "Task '%s' on quantum %s exited early with no work found: %s.",
582 task_node.label,
583 quantum.dataId,
584 str(err),
585 )
586 flags |= err.FLAGS
587 except AnnotatedPartialOutputsError as caught:
588 error: BaseException
589 if caught.__cause__ is None:
590 _LOG.error(
591 "Incorrect use of AnnotatedPartialOutputsError: no chained exception found.",
592 task_node.label,
593 quantum.dataId,
594 )
595 error = caught
596 else:
597 error = caught.__cause__
598 if self._raise_on_partial_outputs:
599 # Note: this is a real edge case that required some
600 # experimentation: without 'from None' below, this raise would
601 # produce a "while one exception was being handled, another was
602 # raised" traceback involving AnnotatedPartialOutputsError.
603 # With the 'from None', we get just the error chained to it, as
604 # desired.
605 raise error from None
606 else:
607 _LOG.error(
608 "Task '%s' on quantum %s exited with partial outputs; "
609 "considering this a qualified success and proceeding.",
610 task_node.label,
611 quantum.dataId,
612 )
613 _LOG.error(error, exc_info=error)
614 flags |= caught.FLAGS
615 finally:
616 ids_put.extend(output[2] for output in butlerQC.outputsPut)
617 if not butlerQC.outputsPut:
618 flags |= QuantumSuccessCaveats.ALL_OUTPUTS_MISSING
619 if not butlerQC.outputsPut == butlerQC.allOutputs:
620 flags |= QuantumSuccessCaveats.ANY_OUTPUTS_MISSING
621 return flags
623 def _write_metadata(
624 self, quantum: Quantum, metadata: Any, task_node: TaskNode, /, limited_butler: LimitedButler
625 ) -> None:
626 # DatasetRef has to be in the Quantum outputs, can lookup by name
627 try:
628 [ref] = quantum.outputs[task_node.metadata_output.dataset_type_name]
629 except LookupError as exc:
630 raise InvalidQuantumError(
631 "Quantum outputs is missing metadata dataset type "
632 f"{task_node.metadata_output.dataset_type_name};"
633 " this could happen due to inconsistent options between QuantumGraph generation"
634 " and execution"
635 ) from exc
636 limited_butler.put(metadata, ref)
638 def _should_assume_exists(self, quantum: Quantum, ref: DatasetRef) -> bool | None:
639 """Report whether the given dataset can be assumed to exist because
640 some previous check reported that it did.
642 If this is `True` for a dataset does not in fact exist anymore, that's
643 an unexpected problem that we want to raise as an exception, and
644 definitely not a case where some predicted output just wasn't produced.
645 We can't always tell the difference, but in this case we can.
647 Parameters
648 ----------
649 quantum : `Quantum`
650 Quantum being processed.
651 ref : `lsst.daf.butler.DatasetRef`
652 Reference to the input dataset.
654 Returns
655 -------
656 exists : `bool` or `None`
657 `True` if this dataset is definitely an overall input, `False` if
658 some other quantum in the graph is expected to produce it, and
659 `None` if the answer could not be determined.
660 """
661 if quantum.datastore_records:
662 for datastore_record_data in quantum.datastore_records.values():
663 if ref.id in datastore_record_data.records:
664 return True
665 return False
666 return None