Coverage for python / lsst / daf / butler / _labeled_butler_factory.py: 34%
78 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-12-19 01:11 -0800
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 typing import Protocol
35from lsst.resources import ResourcePathExpression
37from ._butler import Butler
38from ._butler_config import ButlerConfig, ButlerType
39from ._butler_repo_index import ButlerRepoIndex
40from ._utilities.named_locks import NamedLocks
41from ._utilities.thread_safe_cache import ThreadSafeCache
44class LabeledButlerFactoryProtocol(Protocol):
45 """Callable to retrieve a butler from a label."""
47 def __call__(self, label: str) -> Butler: ... 47 ↛ exitline 47 didn't return from function '__call__' because
50class LabeledButlerFactory:
51 """Factory for efficiently instantiating Butler instances from the
52 repository index file. This is intended for use from long-lived services
53 that want to instantiate a separate Butler instance for each end user
54 request.
56 Parameters
57 ----------
58 repositories : `~collections.abc.Mapping` [`str`, `str`], optional
59 Keys are arbitrary labels, and values are URIs to Butler configuration
60 files. If not provided, defaults to the global repository index
61 configured by the ``DAF_BUTLER_REPOSITORY_INDEX`` environment variable
62 -- see `ButlerRepoIndex`.
64 Notes
65 -----
66 This interface is currently considered experimental and is subject to
67 change.
69 For each label in the repository index, caches shared state to allow fast
70 instantiation of new instances.
72 Instance methods on this class are threadsafe -- a single instance of
73 `LabeledButlerFactory` can be used concurrently by multiple threads. It is
74 NOT safe for a single `Butler` instance returned by this factory to be used
75 concurrently by multiple threads. However, separate `Butler` instances can
76 safely be used by separate threads.
77 """
79 def __init__(self, repositories: Mapping[str, str] | None = None) -> None:
80 if repositories is None:
81 self._repositories = None
82 else:
83 self._repositories = dict(repositories)
85 self._factories = ThreadSafeCache[str, _ButlerFactory]()
86 self._initialization_locks = NamedLocks()
88 # This may be overridden by unit tests.
89 self._preload_unsafe_direct_butler_caches = True
91 def bind(self, access_token: str | None) -> LabeledButlerFactoryProtocol:
92 """Create a callable factory function for generating Butler instances
93 with out needing to specify access tokans again.
95 Parameters
96 ----------
97 access_token : `str` or `None`
98 An optional access token to use for authentication with the Butler.
100 Returns
101 -------
102 bound : `LabeledButlerFactoryProtocol`
103 A callable that takes a label as input and returns a Butler
104 instance.
105 """
107 def create(label: str) -> Butler:
108 return self.create_butler(label=label, access_token=access_token)
110 return create
112 def create_butler(self, *, label: str, access_token: str | None) -> Butler:
113 """Create a Butler instance.
115 Parameters
116 ----------
117 label : `str`
118 Label of the repository to instantiate, from the ``repositories``
119 parameter to the `LabeledButlerFactory` constructor or the global
120 repository index file.
121 access_token : `str` | `None`
122 Gafaelfawr access token used to authenticate to a Butler server.
123 This is required for any repositories configured to use
124 `RemoteButler`. If you only use `DirectButler`, this may be
125 `None`.
127 Raises
128 ------
129 KeyError
130 Raised if the label is not found in the index.
132 Notes
133 -----
134 For a service making requests on behalf of end users, the access token
135 should normally be a "delegated" token so that access permissions are
136 based on the end user instead of the service. See
137 https://gafaelfawr.lsst.io/user-guide/gafaelfawringress.html#requesting-delegated-tokens
138 """
139 factory = self._get_or_create_butler_factory(label)
140 return factory.create_butler(access_token)
142 def close(self) -> None:
143 """Reset the factory cache, and release any resources associated with
144 the cached instances.
145 """
146 factories = self._factories.clear()
147 for factory in factories.values():
148 factory.close()
150 def _get_or_create_butler_factory(self, label: str) -> _ButlerFactory:
151 # We maintain a separate lock per label. We only want to instantiate
152 # one factory function per label, because creating the factory sets up
153 # shared state that should only exist once per repository. However, we
154 # don't want other repositories' instance creation to block on one
155 # repository that is slow to initialize.
156 with self._initialization_locks.lock(label):
157 if (factory := self._factories.get(label)) is not None:
158 return factory
160 factory = self._create_butler_factory_function(label)
161 return self._factories.set_or_get(label, factory)
163 def _create_butler_factory_function(self, label: str) -> _ButlerFactory:
164 config_uri = self._get_config_uri(label)
165 config = ButlerConfig(config_uri)
166 butler_type = config.get_butler_type()
168 match butler_type:
169 case ButlerType.DIRECT:
170 return _DirectButlerFactory(config, self._preload_unsafe_direct_butler_caches)
171 case ButlerType.REMOTE:
172 return _RemoteButlerFactory(config)
173 case _:
174 raise TypeError(f"Unknown butler type '{butler_type}' for label '{label}'")
176 def _get_config_uri(self, label: str) -> ResourcePathExpression:
177 if self._repositories is None:
178 return ButlerRepoIndex.get_repo_uri(label)
179 else:
180 config_uri = self._repositories.get(label)
181 if config_uri is None:
182 raise KeyError(f"Unknown repository label '{label}'")
183 return config_uri
186class _ButlerFactory(Protocol):
187 def create_butler(self, access_token: str | None) -> Butler: ... 187 ↛ exitline 187 didn't return from function 'create_butler' because
188 def close(self) -> None: ... 188 ↛ exitline 188 didn't return from function 'close' because
191class _DirectButlerFactory(_ButlerFactory):
192 def __init__(self, config: ButlerConfig, preload_unsafe_caches: bool) -> None:
193 import lsst.daf.butler.direct_butler
195 # Create a 'template' Butler that will be cloned when callers request
196 # an instance.
197 self._butler = Butler.from_config(config)
198 assert isinstance(self._butler, lsst.daf.butler.direct_butler.DirectButler)
200 # Load caches so that data is available in cloned instances without
201 # needing to refetch it from the database for every instance.
202 self._butler._preload_cache(load_dimension_record_cache=preload_unsafe_caches)
204 def create_butler(self, access_token: str | None) -> Butler:
205 # Access token is ignored because DirectButler does not use Gafaelfawr
206 # authentication.
207 return self._butler.clone()
209 def close(self) -> None:
210 self._butler.close()
213class _RemoteButlerFactory(_ButlerFactory):
214 def __init__(self, config: ButlerConfig) -> None:
215 import lsst.daf.butler.remote_butler._factory
217 self._factory = lsst.daf.butler.remote_butler._factory.RemoteButlerFactory.create_factory_from_config(
218 config
219 )
221 def create_butler(self, access_token: str | None) -> Butler:
222 if access_token is None:
223 raise ValueError("Access token is required to connect to a Butler server")
224 return self._factory.create_butler_for_access_token(access_token)
226 def close(self) -> None:
227 pass