Coverage for python / lsst / daf / butler / _labeled_butler_factory.py: 35%
90 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-28 08:36 +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__ = ("LabeledButlerFactory", "LabeledButlerFactoryProtocol")
32from collections.abc import Mapping
33from contextlib import AbstractContextManager
34from logging import getLogger
35from typing import Any, Literal, Protocol, Self
37from lsst.resources import ResourcePathExpression
39from ._butler import Butler
40from ._butler_config import ButlerConfig, ButlerType
41from ._butler_repo_index import ButlerRepoIndex
42from ._utilities.named_locks import NamedLocks
43from ._utilities.thread_safe_cache import ThreadSafeCache
45_LOG = getLogger(__name__)
48class LabeledButlerFactoryProtocol(Protocol):
49 """Callable to retrieve a butler from a label."""
51 def __call__(self, label: str) -> Butler: ... 51 ↛ exitline 51 didn't return from function '__call__' because
54class LabeledButlerFactory(AbstractContextManager):
55 """Factory for efficiently instantiating Butler instances from the
56 repository index file. This is intended for use from long-lived services
57 that want to instantiate a separate Butler instance for each end user
58 request.
60 Parameters
61 ----------
62 repositories : `~collections.abc.Mapping` [`str`, `str`], optional
63 Keys are arbitrary labels, and values are URIs to Butler configuration
64 files. If not provided, defaults to the global repository index
65 configured by the ``DAF_BUTLER_REPOSITORY_INDEX`` environment variable
66 -- see `ButlerRepoIndex`.
67 writeable : `bool`, optional
68 If `True`, Butler instances created by this factory will be writeable.
69 If `False` (the default), instances will be read-only.
71 Notes
72 -----
73 This interface is currently considered experimental and is subject to
74 change.
76 For each label in the repository index, caches shared state to allow fast
77 instantiation of new instances.
79 Instance methods on this class are threadsafe -- a single instance of
80 `LabeledButlerFactory` can be used concurrently by multiple threads. It is
81 NOT safe for a single `Butler` instance returned by this factory to be used
82 concurrently by multiple threads. However, separate `Butler` instances can
83 safely be used by separate threads.
84 """
86 def __init__(self, repositories: Mapping[str, str] | None = None, writeable: bool = False) -> None:
87 if repositories is None:
88 self._repositories = None
89 else:
90 self._repositories = dict(repositories)
91 self._writeable = writeable
93 self._factories = ThreadSafeCache[str, _ButlerFactory]()
94 self._initialization_locks = NamedLocks()
96 # This may be overridden by unit tests.
97 self._preload_unsafe_direct_butler_caches = True
99 def __enter__(self) -> Self:
100 return self
102 def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> Literal[False]:
103 try:
104 self.close()
105 except Exception:
106 _LOG.exception("An exception occurred during LabeledButlerFactory.close()")
107 return False
109 def bind(self, access_token: str | None) -> LabeledButlerFactoryProtocol:
110 """Create a callable factory function for generating Butler instances
111 with out needing to specify access tokans again.
113 Parameters
114 ----------
115 access_token : `str` or `None`
116 An optional access token to use for authentication with the Butler.
118 Returns
119 -------
120 bound : `LabeledButlerFactoryProtocol`
121 A callable that takes a label as input and returns a Butler
122 instance.
123 """
125 def create(label: str) -> Butler:
126 return self.create_butler(label=label, access_token=access_token)
128 return create
130 def create_butler(self, label: str, *, access_token: str | None = None) -> Butler:
131 """Create a Butler instance.
133 Parameters
134 ----------
135 label : `str`
136 Label of the repository to instantiate, from the ``repositories``
137 parameter to the `LabeledButlerFactory` constructor or the global
138 repository index file.
139 access_token : `str` | `None`, optional
140 Gafaelfawr access token used to authenticate to a Butler server.
141 This is required for any repositories configured to use
142 `RemoteButler`. If you only use `DirectButler`, this may be
143 `None`.
145 Raises
146 ------
147 KeyError
148 Raised if the label is not found in the index.
150 Notes
151 -----
152 For a service making requests on behalf of end users, the access token
153 should normally be a "delegated" token so that access permissions are
154 based on the end user instead of the service. See
155 https://gafaelfawr.lsst.io/user-guide/gafaelfawringress.html#requesting-delegated-tokens
156 """
157 factory = self._get_or_create_butler_factory(label)
158 return factory.create_butler(access_token)
160 def close(self) -> None:
161 """Reset the factory cache, and release any resources associated with
162 the cached instances.
163 """
164 factories = self._factories.clear()
165 for factory in factories.values():
166 factory.close()
168 def _get_or_create_butler_factory(self, label: str) -> _ButlerFactory:
169 # We maintain a separate lock per label. We only want to instantiate
170 # one factory function per label, because creating the factory sets up
171 # shared state that should only exist once per repository. However, we
172 # don't want other repositories' instance creation to block on one
173 # repository that is slow to initialize.
174 with self._initialization_locks.lock(label):
175 if (factory := self._factories.get(label)) is not None:
176 return factory
178 factory = self._create_butler_factory_function(label)
179 return self._factories.set_or_get(label, factory)
181 def _create_butler_factory_function(self, label: str) -> _ButlerFactory:
182 config_uri = self._get_config_uri(label)
183 config = ButlerConfig(config_uri)
184 butler_type = config.get_butler_type()
186 match butler_type:
187 case ButlerType.DIRECT:
188 return _DirectButlerFactory(
189 config, self._preload_unsafe_direct_butler_caches, self._writeable
190 )
191 case ButlerType.REMOTE:
192 return _RemoteButlerFactory(config)
193 case _:
194 raise TypeError(f"Unknown butler type '{butler_type}' for label '{label}'")
196 def _get_config_uri(self, label: str) -> ResourcePathExpression:
197 if self._repositories is None:
198 return ButlerRepoIndex.get_repo_uri(label)
199 else:
200 config_uri = self._repositories.get(label)
201 if config_uri is None:
202 raise KeyError(f"Unknown repository label '{label}'")
203 return config_uri
206class _ButlerFactory(Protocol):
207 def create_butler(self, access_token: str | None) -> Butler: ... 207 ↛ exitline 207 didn't return from function 'create_butler' because
208 def close(self) -> None: ... 208 ↛ exitline 208 didn't return from function 'close' because
211class _DirectButlerFactory(_ButlerFactory):
212 def __init__(self, config: ButlerConfig, preload_unsafe_caches: bool, writeable: bool) -> None:
213 import lsst.daf.butler.direct_butler
215 # Create a 'template' Butler that will be cloned when callers request
216 # an instance.
217 self._butler = Butler.from_config(config, writeable=writeable)
218 assert isinstance(self._butler, lsst.daf.butler.direct_butler.DirectButler)
220 # Load caches so that data is available in cloned instances without
221 # needing to refetch it from the database for every instance.
222 self._butler._preload_cache(load_dimension_record_cache=preload_unsafe_caches)
224 def create_butler(self, access_token: str | None) -> Butler:
225 # Access token is ignored because DirectButler does not use Gafaelfawr
226 # authentication.
227 return self._butler.clone()
229 def close(self) -> None:
230 self._butler.close()
233class _RemoteButlerFactory(_ButlerFactory):
234 def __init__(self, config: ButlerConfig) -> None:
235 import lsst.daf.butler.remote_butler._factory
237 self._factory = lsst.daf.butler.remote_butler._factory.RemoteButlerFactory.create_factory_from_config(
238 config
239 )
241 def create_butler(self, access_token: str | None) -> Butler:
242 if access_token is None:
243 raise ValueError("Access token is required to connect to a Butler server")
244 return self._factory.create_butler_for_access_token(access_token)
246 def close(self) -> None:
247 pass