Coverage for python/lsst/daf/butler/remote_butler/_remote_butler.py: 31%
275 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-10-10 01:28 -0700
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__ = ("RemoteButler",)
32import logging
33import uuid
34from collections.abc import Collection, Iterable, Iterator, Sequence
35from contextlib import AbstractContextManager, contextmanager
36from types import EllipsisType
37from typing import TYPE_CHECKING, Any, TextIO, cast
39from deprecated.sphinx import deprecated
41from lsst.daf.butler.datastores.file_datastore.retrieve_artifacts import (
42 ArtifactIndexInfo,
43 ZipIndex,
44 determine_destination_for_retrieved_artifact,
45 retrieve_and_zip,
46 unpack_zips,
47)
48from lsst.resources import ResourcePath, ResourcePathExpression
50from .._butler import Butler, _DeprecatedDefault
51from .._butler_collections import ButlerCollections
52from .._butler_metrics import ButlerMetrics
53from .._dataset_existence import DatasetExistence
54from .._dataset_ref import DatasetId, DatasetRef
55from .._dataset_type import DatasetType
56from .._deferredDatasetHandle import DeferredDatasetHandle
57from .._exceptions import DatasetNotFoundError
58from .._query_all_datasets import QueryAllDatasetsParameters
59from .._storage_class import StorageClass, StorageClassFactory
60from .._utilities.locked_object import LockedObject
61from ..datastore import DatasetRefURIs, DatastoreConfig
62from ..datastore.cache_manager import AbstractDatastoreCacheManager, DatastoreCacheManager
63from ..dimensions import DataCoordinate, DataIdValue, DimensionConfig, DimensionUniverse, SerializedDataId
64from ..queries import Query
65from ..queries.tree import make_column_literal
66from ..registry import CollectionArgType, NoDefaultCollectionError, Registry, RegistryDefaults
67from ._collection_args import convert_collection_arg_to_glob_string_list
68from ._defaults import DefaultsHolder
69from ._get import convert_http_url_to_resource_path, get_dataset_as_python_object
70from ._http_connection import RemoteButlerHttpConnection, parse_model, quote_path_variable
71from ._query_driver import RemoteQueryDriver
72from ._query_results import convert_dataset_ref_results, read_query_results
73from ._ref_utils import apply_storage_class_override, normalize_dataset_type_name, simplify_dataId
74from ._registry import RemoteButlerRegistry
75from ._remote_butler_collections import RemoteButlerCollections
76from ._remote_file_transfer_source import RemoteFileTransferSource
77from .server_models import (
78 CollectionList,
79 FileInfoPayload,
80 FindDatasetRequestModel,
81 FindDatasetResponseModel,
82 GetDatasetTypeResponseModel,
83 GetFileByDataIdRequestModel,
84 GetFileResponseModel,
85 GetUniverseResponseModel,
86 QueryAllDatasetsRequestModel,
87)
89if TYPE_CHECKING:
90 from .._dataset_provenance import DatasetProvenance
91 from .._file_dataset import FileDataset
92 from .._limited_butler import LimitedButler
93 from .._timespan import Timespan
94 from ..dimensions import DataId
95 from ..transfers import RepoExportContext
98_LOG = logging.getLogger(__name__)
101class RemoteButler(Butler): # numpydoc ignore=PR02
102 """A `Butler` that can be used to connect through a remote server.
104 Parameters
105 ----------
106 options : `ButlerInstanceOptions`
107 Default values and other settings for the Butler instance.
108 connection : `RemoteButlerHttpConnection`
109 Connection to Butler server.
110 cache : `RemoteButlerCache`
111 Cache of data shared between multiple RemoteButler instances connected
112 to the same server.
113 use_disabled_datastore_cache : `bool`, optional
114 If `True`, a datastore cache manager will be created with a default
115 disabled state which can be enabled by the environment. If `False`
116 a cache manager will be constructed from the default local
117 configuration, likely caching by default but only specific storage
118 classes.
120 Notes
121 -----
122 Instead of using this constructor, most users should use either
123 `Butler.from_config` or `RemoteButlerFactory`.
124 """
126 _registry_defaults: DefaultsHolder
127 _connection: RemoteButlerHttpConnection
128 _cache: RemoteButlerCache
129 _registry: RemoteButlerRegistry
130 _datastore_cache_manager: AbstractDatastoreCacheManager | None
131 _use_disabled_datastore_cache: bool
133 # This is __new__ instead of __init__ because we have to support
134 # instantiation via the legacy constructor Butler.__new__(), which
135 # reads the configuration and selects which subclass to instantiate. The
136 # interaction between __new__ and __init__ is kind of wacky in Python. If
137 # we were using __init__ here, __init__ would be called twice (once when
138 # the RemoteButler instance is constructed inside Butler.from_config(), and
139 # a second time with the original arguments to Butler() when the instance
140 # is returned from Butler.__new__()
141 def __new__(
142 cls,
143 *,
144 connection: RemoteButlerHttpConnection,
145 defaults: RegistryDefaults,
146 cache: RemoteButlerCache,
147 use_disabled_datastore_cache: bool = True,
148 metrics: ButlerMetrics | None = None,
149 ) -> RemoteButler:
150 self = cast(RemoteButler, super().__new__(cls))
151 self.storageClasses = StorageClassFactory()
153 self._connection = connection
154 self._cache = cache
155 self._datastore_cache_manager = None
156 self._use_disabled_datastore_cache = use_disabled_datastore_cache
157 self._metrics = metrics if metrics is not None else ButlerMetrics()
159 self._registry_defaults = DefaultsHolder(defaults)
160 self._registry = RemoteButlerRegistry(self, self._registry_defaults, self._connection)
161 defaults.finish(self._registry)
163 return self
165 def isWriteable(self) -> bool:
166 # Docstring inherited.
167 return False
169 @property
170 @deprecated(
171 "Please use 'collections' instead. collection_chains will be removed after v28.",
172 version="v28",
173 category=FutureWarning,
174 )
175 def collection_chains(self) -> ButlerCollections:
176 """Object with methods for modifying collection chains."""
177 return self.collections
179 @property
180 def collections(self) -> ButlerCollections:
181 """Object with methods for modifying and querying collections."""
182 return RemoteButlerCollections(self._registry_defaults, self._connection)
184 @property
185 def dimensions(self) -> DimensionUniverse:
186 # Docstring inherited.
187 with self._cache.access() as cache:
188 if cache.dimensions is not None:
189 return cache.dimensions
191 response = self._connection.get("universe")
192 model = parse_model(response, GetUniverseResponseModel)
194 config = DimensionConfig.from_simple(model.universe)
195 universe = DimensionUniverse(config)
196 with self._cache.access() as cache:
197 if cache.dimensions is None:
198 cache.dimensions = universe
199 return cache.dimensions
201 @property
202 def _cache_manager(self) -> AbstractDatastoreCacheManager:
203 """Cache manager to use when reading files from the butler."""
204 # RemoteButler does not get any cache configuration from the server.
205 # Either create a disabled cache manager which can be enabled via the
206 # environment, or create a cache manager from the default FileDatastore
207 # config. This will not work properly if the defaults for
208 # DatastoreConfig no longer include the cache.
209 if self._datastore_cache_manager is None:
210 datastore_config = DatastoreConfig()
211 if not self._use_disabled_datastore_cache and "cached" in datastore_config:
212 self._datastore_cache_manager = DatastoreCacheManager(
213 datastore_config["cached"], universe=self.dimensions
214 )
215 else:
216 self._datastore_cache_manager = DatastoreCacheManager.create_disabled(
217 universe=self.dimensions
218 )
219 return self._datastore_cache_manager
221 def _caching_context(self) -> AbstractContextManager[None]:
222 # Docstring inherited.
223 # Not implemented for now, will have to think whether this needs to
224 # do something on client side and/or remote side.
225 raise NotImplementedError()
227 def transaction(self) -> AbstractContextManager[None]:
228 """Will always raise NotImplementedError.
229 Transactions are not supported by RemoteButler.
230 """
231 raise NotImplementedError()
233 def put(
234 self,
235 obj: Any,
236 datasetRefOrType: DatasetRef | DatasetType | str,
237 /,
238 dataId: DataId | None = None,
239 *,
240 run: str | None = None,
241 provenance: DatasetProvenance | None = None,
242 **kwargs: Any,
243 ) -> DatasetRef:
244 # Docstring inherited.
245 raise NotImplementedError()
247 def getDeferred(
248 self,
249 datasetRefOrType: DatasetRef | DatasetType | str,
250 /,
251 dataId: DataId | None = None,
252 *,
253 parameters: dict | None = None,
254 collections: Any = None,
255 storageClass: str | StorageClass | None = None,
256 timespan: Timespan | None = None,
257 **kwargs: Any,
258 ) -> DeferredDatasetHandle:
259 response = self._get_file_info(datasetRefOrType, dataId, collections, timespan, kwargs)
260 # Check that artifact information is available.
261 _to_file_payload(response)
262 ref = DatasetRef.from_simple(response.dataset_ref, universe=self.dimensions)
263 return DeferredDatasetHandle(butler=self, ref=ref, parameters=parameters, storageClass=storageClass)
265 def get(
266 self,
267 datasetRefOrType: DatasetRef | DatasetType | str,
268 /,
269 dataId: DataId | None = None,
270 *,
271 parameters: dict[str, Any] | None = None,
272 collections: Any = None,
273 storageClass: StorageClass | str | None = None,
274 timespan: Timespan | None = None,
275 **kwargs: Any,
276 ) -> Any:
277 # Docstring inherited.
278 with self._metrics.instrument_get(log=_LOG, msg="Retrieved remote dataset"):
279 model = self._get_file_info(datasetRefOrType, dataId, collections, timespan, kwargs)
281 ref = DatasetRef.from_simple(model.dataset_ref, universe=self.dimensions)
282 # If the caller provided a DatasetRef, they may have overridden the
283 # component on it. We need to explicitly handle this because we
284 # did not send the DatasetType to the server in this case.
285 if isinstance(datasetRefOrType, DatasetRef):
286 componentOverride = datasetRefOrType.datasetType.component()
287 if componentOverride:
288 ref = ref.makeComponentRef(componentOverride)
289 ref = apply_storage_class_override(ref, datasetRefOrType, storageClass)
291 return self._get_dataset_as_python_object(ref, model, parameters)
293 def _get_dataset_as_python_object(
294 self,
295 ref: DatasetRef,
296 model: GetFileResponseModel,
297 parameters: dict[str, Any] | None,
298 ) -> Any:
299 # This thin wrapper method is here to provide a place to hook in a mock
300 # mimicking DatastoreMock functionality for use in unit tests.
301 return get_dataset_as_python_object(
302 ref,
303 _to_file_payload(model),
304 auth=self._connection.auth,
305 parameters=parameters,
306 cache_manager=self._cache_manager,
307 )
309 def _get_file_info(
310 self,
311 datasetRefOrType: DatasetRef | DatasetType | str,
312 dataId: DataId | None,
313 collections: CollectionArgType,
314 timespan: Timespan | None,
315 kwargs: dict[str, DataIdValue],
316 ) -> GetFileResponseModel:
317 """Send a request to the server for the file URLs and metadata
318 associated with a dataset.
319 """
320 if isinstance(datasetRefOrType, DatasetRef):
321 if dataId is not None:
322 raise ValueError("DatasetRef given, cannot use dataId as well")
323 return self._get_file_info_for_ref(datasetRefOrType)
324 else:
325 request = GetFileByDataIdRequestModel(
326 dataset_type=normalize_dataset_type_name(datasetRefOrType),
327 collections=self._normalize_collections(collections),
328 data_id=simplify_dataId(dataId, kwargs),
329 default_data_id=self._serialize_default_data_id(),
330 timespan=timespan,
331 )
332 response = self._connection.post("get_file_by_data_id", request)
333 return parse_model(response, GetFileResponseModel)
335 def _get_file_info_for_ref(self, ref: DatasetRef) -> GetFileResponseModel:
336 response = self._connection.get(f"get_file/{_to_uuid_string(ref.id)}")
337 return parse_model(response, GetFileResponseModel)
339 def getURIs(
340 self,
341 datasetRefOrType: DatasetRef | DatasetType | str,
342 /,
343 dataId: DataId | None = None,
344 *,
345 predict: bool = False,
346 collections: Any = None,
347 run: str | None = None,
348 **kwargs: Any,
349 ) -> DatasetRefURIs:
350 # Docstring inherited.
351 if predict or run:
352 raise NotImplementedError("Predict mode is not supported by RemoteButler")
354 response = self._get_file_info(datasetRefOrType, dataId, collections, None, kwargs)
355 file_info = _to_file_payload(response).file_info
356 if len(file_info) == 1:
357 return DatasetRefURIs(
358 primaryURI=convert_http_url_to_resource_path(
359 file_info[0].url, self._connection.auth, file_info[0].auth
360 )
361 )
362 else:
363 components = {}
364 for f in file_info:
365 component = f.datastoreRecords.component
366 if component is None:
367 raise ValueError(
368 f"DatasetId {response.dataset_ref.id} has a component file"
369 " with no component name defined"
370 )
371 components[component] = convert_http_url_to_resource_path(
372 f.url, self._connection.auth, f.auth
373 )
374 return DatasetRefURIs(componentURIs=components)
376 def get_dataset_type(self, name: str) -> DatasetType:
377 with self._cache.access() as cache:
378 if (cached_value := cache.dataset_types.get(name)) is not None:
379 return cached_value
381 response = self._connection.get(f"dataset_type/{quote_path_variable(name)}")
382 model = parse_model(response, GetDatasetTypeResponseModel)
383 value = DatasetType.from_simple(model.dataset_type, universe=self.dimensions)
384 with self._cache.access() as cache:
385 return cache.dataset_types.setdefault(name, value)
387 def get_dataset(
388 self,
389 id: DatasetId,
390 *,
391 storage_class: str | StorageClass | None = None,
392 dimension_records: bool = False,
393 datastore_records: bool = False,
394 ) -> DatasetRef | None:
395 # datastore_records is intentionally ignored. It is an optimization
396 # flag that only applies to DirectButler.
397 path = f"dataset/{_to_uuid_string(id)}"
398 response = self._connection.get(path, params={"dimension_records": bool(dimension_records)})
399 model = parse_model(response, FindDatasetResponseModel)
400 if model.dataset_ref is None:
401 return None
402 ref = DatasetRef.from_simple(model.dataset_ref, universe=self.dimensions)
403 if storage_class is not None:
404 ref = ref.overrideStorageClass(storage_class)
405 return ref
407 def find_dataset(
408 self,
409 dataset_type: DatasetType | str,
410 data_id: DataId | None = None,
411 *,
412 collections: str | Sequence[str] | None = None,
413 timespan: Timespan | None = None,
414 storage_class: str | StorageClass | None = None,
415 dimension_records: bool = False,
416 datastore_records: bool = False,
417 **kwargs: Any,
418 ) -> DatasetRef | None:
419 # datastore_records is intentionally ignored. It is an optimization
420 # flag that only applies to DirectButler.
422 query = FindDatasetRequestModel(
423 dataset_type=normalize_dataset_type_name(dataset_type),
424 data_id=simplify_dataId(data_id, kwargs),
425 default_data_id=self._serialize_default_data_id(),
426 collections=self._normalize_collections(collections),
427 timespan=timespan,
428 dimension_records=dimension_records,
429 )
431 response = self._connection.post("find_dataset", query)
433 model = parse_model(response, FindDatasetResponseModel)
434 if model.dataset_ref is None:
435 return None
437 ref = DatasetRef.from_simple(model.dataset_ref, universe=self.dimensions)
438 if isinstance(data_id, DataCoordinate) and data_id.hasRecords():
439 ref = ref.expanded(data_id)
440 return apply_storage_class_override(ref, dataset_type, storage_class)
442 def _retrieve_artifacts(
443 self,
444 refs: Iterable[DatasetRef],
445 destination: ResourcePathExpression,
446 transfer: str = "auto",
447 preserve_path: bool = True,
448 overwrite: bool = False,
449 write_index: bool = True,
450 add_prefix: bool = False,
451 ) -> dict[ResourcePath, ArtifactIndexInfo]:
452 destination = ResourcePath(destination).abspath()
453 if not destination.isdir():
454 raise ValueError(f"Destination location must refer to a directory. Given {destination}.")
456 if transfer not in ("auto", "copy"):
457 raise ValueError("Only 'copy' and 'auto' transfer modes are supported.")
459 requested_ids = {ref.id for ref in refs}
460 have_copied: dict[ResourcePath, ResourcePath] = {}
461 artifact_map: dict[ResourcePath, ArtifactIndexInfo] = {}
462 # Sort to ensure that in many refs to one file situation the same
463 # ref is used for any prefix that might be added.
464 for ref in sorted(refs):
465 prefix = str(ref.id)[:8] + "-" if add_prefix else ""
466 file_info = _to_file_payload(self._get_file_info_for_ref(ref)).file_info
467 for file in file_info:
468 source_uri = ResourcePath(str(file.url))
469 # For DECam/zip we only want to copy once.
470 # For zip files we need to unpack so that they can be
471 # zipped up again if needed.
472 is_zip = source_uri.getExtension() == ".zip" and "zip-path" in source_uri.fragment
473 cleaned_source_uri = source_uri.replace(fragment="", query="", params="")
474 if is_zip:
475 if cleaned_source_uri not in have_copied:
476 zipped_artifacts = unpack_zips(
477 [cleaned_source_uri], requested_ids, destination, preserve_path, overwrite
478 )
479 artifact_map.update(zipped_artifacts)
480 have_copied[cleaned_source_uri] = cleaned_source_uri
481 elif cleaned_source_uri not in have_copied:
482 relative_path = ResourcePath(file.datastoreRecords.path, forceAbsolute=False)
483 target_uri = determine_destination_for_retrieved_artifact(
484 destination, relative_path, preserve_path, prefix
485 )
486 # Because signed URLs expire, we want to do the transfer
487 # soon after retrieving the URL.
488 target_uri.transfer_from(source_uri, transfer="copy", overwrite=overwrite)
489 have_copied[cleaned_source_uri] = target_uri
490 artifact_map[target_uri] = ArtifactIndexInfo.from_single(file.datastoreRecords, ref.id)
491 else:
492 target_uri = have_copied[cleaned_source_uri]
493 artifact_map[target_uri].append(ref.id)
495 if write_index:
496 index = ZipIndex.from_artifact_map(refs, artifact_map, destination)
497 index.write_index(destination)
499 return artifact_map
501 def retrieve_artifacts_zip(
502 self,
503 refs: Iterable[DatasetRef],
504 destination: ResourcePathExpression,
505 overwrite: bool = True,
506 ) -> ResourcePath:
507 return retrieve_and_zip(refs, destination, self._retrieve_artifacts, overwrite)
509 def retrieveArtifacts(
510 self,
511 refs: Iterable[DatasetRef],
512 destination: ResourcePathExpression,
513 transfer: str = "auto",
514 preserve_path: bool = True,
515 overwrite: bool = False,
516 ) -> list[ResourcePath]:
517 artifact_map = self._retrieve_artifacts(
518 refs,
519 destination,
520 transfer,
521 preserve_path,
522 overwrite,
523 )
524 return list(artifact_map)
526 def exists(
527 self,
528 dataset_ref_or_type: DatasetRef | DatasetType | str,
529 /,
530 data_id: DataId | None = None,
531 *,
532 full_check: bool = True,
533 collections: Any = None,
534 **kwargs: Any,
535 ) -> DatasetExistence:
536 try:
537 response = self._get_file_info(
538 dataset_ref_or_type, dataId=data_id, collections=collections, timespan=None, kwargs=kwargs
539 )
540 except DatasetNotFoundError:
541 return DatasetExistence.UNRECOGNIZED
543 if response.artifact is None:
544 if full_check:
545 return DatasetExistence.RECORDED
546 else:
547 return DatasetExistence.RECORDED | DatasetExistence._ASSUMED
549 if full_check:
550 for file in response.artifact.file_info:
551 if not ResourcePath(str(file.url)).exists():
552 return DatasetExistence.RECORDED | DatasetExistence.DATASTORE
553 return DatasetExistence.VERIFIED
554 else:
555 return DatasetExistence.KNOWN
557 def _exists_many(
558 self,
559 refs: Iterable[DatasetRef],
560 /,
561 *,
562 full_check: bool = True,
563 ) -> dict[DatasetRef, DatasetExistence]:
564 return {ref: self.exists(ref, full_check=full_check) for ref in refs}
566 def removeRuns(
567 self,
568 names: Iterable[str],
569 unstore: bool | type[_DeprecatedDefault] = _DeprecatedDefault,
570 *,
571 unlink_from_chains: bool = False,
572 ) -> None:
573 # Docstring inherited.
574 raise NotImplementedError()
576 def ingest(
577 self,
578 *datasets: FileDataset,
579 transfer: str | None = "auto",
580 record_validation_info: bool = True,
581 skip_existing: bool = False,
582 ) -> None:
583 # Docstring inherited.
584 raise NotImplementedError()
586 def ingest_zip(
587 self,
588 zip_file: ResourcePathExpression,
589 transfer: str = "auto",
590 *,
591 transfer_dimensions: bool = False,
592 dry_run: bool = False,
593 ) -> None:
594 # Docstring inherited.
595 raise NotImplementedError()
597 def export(
598 self,
599 *,
600 directory: str | None = None,
601 filename: str | None = None,
602 format: str | None = None,
603 transfer: str | None = None,
604 ) -> AbstractContextManager[RepoExportContext]:
605 # Docstring inherited.
606 raise NotImplementedError()
608 def import_(
609 self,
610 *,
611 directory: ResourcePathExpression | None = None,
612 filename: ResourcePathExpression | TextIO | None = None,
613 format: str | None = None,
614 transfer: str | None = None,
615 skip_dimensions: set | None = None,
616 record_validation_info: bool = True,
617 without_datastore: bool = False,
618 ) -> None:
619 # Docstring inherited.
620 raise NotImplementedError()
622 def transfer_dimension_records_from(
623 self, source_butler: LimitedButler | Butler, source_refs: Iterable[DatasetRef]
624 ) -> None:
625 # Docstring inherited.
626 raise NotImplementedError()
628 def transfer_from(
629 self,
630 source_butler: LimitedButler,
631 source_refs: Iterable[DatasetRef],
632 transfer: str = "auto",
633 skip_missing: bool = True,
634 register_dataset_types: bool = False,
635 transfer_dimensions: bool = False,
636 dry_run: bool = False,
637 ) -> Collection[DatasetRef]:
638 # Docstring inherited.
639 raise NotImplementedError()
641 def validateConfiguration(
642 self,
643 logFailures: bool = False,
644 datasetTypeNames: Iterable[str] | None = None,
645 ignore: Iterable[str] | None = None,
646 ) -> None:
647 # Docstring inherited.
648 raise NotImplementedError()
650 @property
651 def run(self) -> str | None:
652 # Docstring inherited.
653 return self._registry_defaults.get().run
655 @property
656 def registry(self) -> Registry:
657 return self._registry
659 @contextmanager
660 def query(self) -> Iterator[Query]:
661 driver = RemoteQueryDriver(self, self._connection)
662 with driver:
663 query = Query(driver)
664 yield query
666 @contextmanager
667 def _query_all_datasets_by_page(
668 self, args: QueryAllDatasetsParameters
669 ) -> Iterator[Iterator[list[DatasetRef]]]:
670 universe = self.dimensions
672 request = QueryAllDatasetsRequestModel(
673 collections=self._normalize_collections(args.collections),
674 name=[normalize_dataset_type_name(name) for name in args.name],
675 find_first=args.find_first,
676 data_id=simplify_dataId(args.data_id, args.kwargs),
677 default_data_id=self._serialize_default_data_id(),
678 where=args.where,
679 bind={k: make_column_literal(v) for k, v in args.bind.items()},
680 limit=args.limit,
681 with_dimension_records=args.with_dimension_records,
682 )
683 with self._connection.post_with_stream_response("query/all_datasets", request) as response:
684 pages = read_query_results(response)
685 yield (convert_dataset_ref_results(page, universe) for page in pages)
687 def pruneDatasets(
688 self,
689 refs: Iterable[DatasetRef],
690 *,
691 disassociate: bool = True,
692 unstore: bool = False,
693 tags: Iterable[str] = (),
694 purge: bool = False,
695 ) -> None:
696 # Docstring inherited.
697 raise NotImplementedError()
699 def _normalize_collections(self, collections: CollectionArgType | None) -> CollectionList:
700 """Convert the ``collections`` parameter in the format used by Butler
701 methods to a standardized format for the REST API.
702 """
703 if collections is None:
704 if not self.collections.defaults:
705 raise NoDefaultCollectionError(
706 "No collections provided, and no defaults from butler construction."
707 )
708 collections = self.collections.defaults
709 return convert_collection_arg_to_glob_string_list(collections)
711 def clone(
712 self,
713 *,
714 collections: CollectionArgType | None | EllipsisType = ...,
715 run: str | None | EllipsisType = ...,
716 inferDefaults: bool | EllipsisType = ...,
717 dataId: dict[str, str] | EllipsisType = ...,
718 metrics: ButlerMetrics | None = None,
719 ) -> RemoteButler:
720 defaults = self._registry_defaults.get().clone(collections, run, inferDefaults, dataId)
721 return RemoteButler(
722 connection=self._connection, cache=self._cache, defaults=defaults, metrics=metrics
723 )
725 @property
726 def _file_transfer_source(self) -> RemoteFileTransferSource:
727 return RemoteFileTransferSource(self._connection)
729 def __str__(self) -> str:
730 return f"RemoteButler({self._connection.server_url})"
732 def _serialize_default_data_id(self) -> SerializedDataId:
733 """Convert the default data ID to a serializable format."""
734 # In an ideal world, the default data ID would just get combined with
735 # the rest of the data ID on the client side instead of being sent
736 # separately to the server. Unfortunately, that requires knowledge of
737 # the DatasetType's dimensions which we don't always have available on
738 # the client. Data ID values can be specified indirectly by "implied"
739 # dimensions, but knowing what things are implied depends on what the
740 # required dimensions are.
742 return self._registry_defaults.get().dataId.to_simple(minimal=True).dataId
745def _to_file_payload(get_file_response: GetFileResponseModel) -> FileInfoPayload:
746 if get_file_response.artifact is None:
747 ref = get_file_response.dataset_ref
748 raise DatasetNotFoundError(f"Dataset is known, but artifact is not available. (datasetId='{ref.id}')")
750 return get_file_response.artifact
753def _to_uuid_string(id: uuid.UUID | str) -> str:
754 """Convert a UUID, or string parseable as a UUID, into a string formatted
755 like '1481269e-4c8d-4696-bcca-d1b4c9005d06'
756 """
757 return str(uuid.UUID(str(id)))
760class _RemoteButlerCacheData:
761 def __init__(self) -> None:
762 self.dimensions: DimensionUniverse | None = None
763 self.dataset_types: dict[str, DatasetType] = {}
766class RemoteButlerCache(LockedObject[_RemoteButlerCacheData]):
767 def __init__(self) -> None:
768 super().__init__(_RemoteButlerCacheData())