Coverage for python / lsst / daf / butler / cli / butler.py: 35%
169 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-12-12 01:12 -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/>.
27from __future__ import annotations
29__all__ = (
30 "ButlerCLI",
31 "LoaderCLI",
32 "cli",
33 "main",
34)
36import abc
37import dataclasses
38import functools
39import logging
40import os
41import traceback
42import types
43from collections import defaultdict
44from functools import cache
45from importlib.metadata import entry_points
46from typing import Any
48import click
49import yaml
51from lsst.resources import ResourcePath
52from lsst.utils import doImport
53from lsst.utils.introspection import get_full_type_name
54from lsst.utils.timer import time_this
56from .cliLog import CliLog
57from .opt import log_file_option, log_label_option, log_level_option, log_tty_option, long_log_option
58from .progress import ClickProgressHandler
60log = logging.getLogger(__name__)
63@functools.lru_cache
64def _importPlugin(pluginName: str) -> types.ModuleType | type | None | click.Command:
65 """Import a plugin that contains Click commands.
67 Parameters
68 ----------
69 pluginName : `str`
70 An importable module whose __all__ parameter contains the commands
71 that can be called.
73 Returns
74 -------
75 An imported module or None
76 The imported module, or None if the module could not be imported.
78 Notes
79 -----
80 A cache is used in order to prevent repeated reports of failure
81 to import a module that can be triggered by ``butler --help``.
82 """
83 try:
84 return doImport(pluginName)
85 except Exception as err:
86 log.warning("Could not import plugin from %s, skipping.", pluginName)
87 log.debug(
88 "Plugin import exception: %s\nTraceback:\n%s",
89 err,
90 "".join(traceback.format_tb(err.__traceback__)),
91 )
92 return None
95@dataclasses.dataclass(frozen=True)
96class PluginCommand:
97 """A click Command and the plugin it came from."""
99 command: click.Command
100 """The command (`click.Command`)."""
101 source: str
102 """Where the command came from (`str`)."""
105class LoaderCLI(click.Group, abc.ABC):
106 """Extends `click.MultiCommand`, which dispatches to subcommands, to load
107 subcommands at runtime.
109 Parameters
110 ----------
111 *args : `typing.Any`
112 Arguments passed to parent constructor.
113 **kwargs : `typing.Any`
114 Keyword arguments passed to parent constructor.
115 """
117 def __init__(self, *args: Any, **kwargs: Any) -> None:
118 super().__init__(*args, **kwargs)
120 @property
121 @abc.abstractmethod
122 def localCmdPkg(self) -> str:
123 """Identifies the location of the commands that are in this
124 package.
126 `getLocalCommands` assumes that the commands can be found in
127 `localCmdPkg.__all__`, if this is not the case then getLocalCommands
128 should be overridden.
130 Returns
131 -------
132 package : `str`
133 The fully qualified location of this package.
134 """
135 raise NotImplementedError()
137 def getLocalCommands(self) -> defaultdict[str, list[str | PluginCommand]]:
138 """Get the commands offered by the local package. This assumes that the
139 commands can be found in `localCmdPkg.__all__`, if this is not the case
140 then this function should be overridden.
142 Returns
143 -------
144 commands : `defaultdict` [`str`, `list` [`str` | `PluginCommand` ]]
145 The key is the command name. The value is a list of package(s) that
146 contains the command.
147 """
148 commandsLocation = _importPlugin(self.localCmdPkg)
149 if commandsLocation is None:
150 # _importPlugins logs an error, don't need to do it again here.
151 return defaultdict(list)
152 assert hasattr(commandsLocation, "__all__"), f"Must define __all__ in {commandsLocation}"
153 commands = [getattr(commandsLocation, name) for name in commandsLocation.__all__]
154 return defaultdict(
155 list,
156 {
157 command.name: [PluginCommand(command, get_full_type_name(commandsLocation))]
158 for command in commands
159 },
160 )
162 def list_commands(self, ctx: click.Context) -> list[str]:
163 """Get all the commands that can be called by the
164 butler command, it is used to generate the --help output.
166 Used by Click.
168 Parameters
169 ----------
170 ctx : `click.Context`
171 The current Click context.
173 Returns
174 -------
175 commands : `list` [`str`]
176 The names of the commands that can be called by the butler command.
177 """
178 self._setupLogging(ctx)
179 commands = self._getCommands()
180 self._raiseIfDuplicateCommands(commands)
181 return sorted(commands)
183 def get_command(self, ctx: click.Context, name: str) -> click.Command | None:
184 """Get a single command for execution.
186 Used by Click.
188 Parameters
189 ----------
190 ctx : `click.Context`
191 The current Click context.
192 name : `str`
193 The name of the command to return.
195 Returns
196 -------
197 command : `click.Command`
198 A Command that wraps a callable command function.
199 """
200 self._setupLogging(ctx)
201 commands = self._getCommands()
202 if name not in commands:
203 return None
204 self._raiseIfDuplicateCommands(commands)
205 command = commands[name][0]
206 if isinstance(command, str):
207 module_str = command + "." + self._cmdNameToFuncName(name)
208 # The click.command decorator returns an instance of a class, which
209 # is something that doImport is not expecting. We add it in as an
210 # option here to appease mypy.
211 with time_this(log, msg="Importing command %s (via %s)", args=(name, module_str)):
212 plugin = _importPlugin(module_str)
213 if not plugin:
214 return None
215 else:
216 plugin = command.command
217 if not isinstance(plugin, click.Command):
218 raise RuntimeError(
219 f"Command {name!r} loaded from {module_str} is not a click Command, is {type(plugin)}"
220 )
221 return plugin
223 def _setupLogging(self, ctx: click.Context | None) -> None:
224 """Init the logging system and config it for the command.
226 Subcommands may further configure the log settings.
227 """
228 if isinstance(ctx, click.Context):
229 CliLog.initLog(
230 longlog=ctx.params.get(long_log_option.name(), False),
231 log_tty=ctx.params.get(log_tty_option.name(), True),
232 log_file=ctx.params.get(log_file_option.name(), ()),
233 log_label=ctx.params.get(log_label_option.name(), ()),
234 )
235 if log_level_option.name() in ctx.params:
236 CliLog.setLogLevels(ctx.params[log_level_option.name()])
237 else:
238 # This works around a bug in sphinx-click, where it passes in the
239 # click.MultiCommand instead of the context.
240 # https://github.com/click-contrib/sphinx-click/issues/70
241 CliLog.initLog(longlog=False)
242 logging.debug(
243 "The passed-in context was not a click.Context, could not determine --long-log or "
244 "--log-level values."
245 )
247 @classmethod
248 def getPluginList(cls) -> list[ResourcePath]:
249 """Get the list of importable yaml files that contain cli data for this
250 command.
252 Returns
253 -------
254 `list` [`lsst.resources.ResourcePath`]
255 The list of files that contain yaml data about a cli plugin.
256 """
257 yaml_files = []
258 if hasattr(cls, "pluginEnvVar"):
259 pluginModules = os.environ.get(cls.pluginEnvVar)
260 if pluginModules:
261 yaml_files.extend([ResourcePath(p) for p in pluginModules.split(":") if p != ""])
263 return yaml_files
265 @classmethod
266 def _funcNameToCmdName(cls, functionName: str) -> str:
267 """Convert function name to the butler command name: change
268 underscores, (used in functions) to dashes (used in commands), and
269 change local-package command names that conflict with python keywords
270 to a legal function name.
271 """
272 return functionName.replace("_", "-")
274 @classmethod
275 def _cmdNameToFuncName(cls, commandName: str) -> str:
276 """Convert butler command name to function name: change dashes (used in
277 commands) to underscores (used in functions), and for local-package
278 commands names that conflict with python keywords, change the local,
279 legal, function name to the command name.
280 """
281 return commandName.replace("-", "_")
283 @staticmethod
284 def _mergeCommandLists(
285 a: defaultdict[str, list[str | PluginCommand]], b: defaultdict[str, list[str | PluginCommand]]
286 ) -> defaultdict[str, list[str | PluginCommand]]:
287 """Combine two dicts whose keys are strings (command name) and values
288 are list of string (the package(s) that provide the named command).
290 Parameters
291 ----------
292 a : `defaultdict` [`str`, `list` [`str` | `PluginCommand` ]]
293 The key is the command name. The value is a list of package(s) that
294 contains the command.
295 b : (same as a)
297 Returns
298 -------
299 commands : `defaultdict` [`str`: [`str` | `PluginCommand` ]]
300 For convenience, returns a extended with b. ('a' is modified in
301 place.)
302 """
303 for key, val in b.items():
304 a[key].extend(val)
305 return a
307 @classmethod
308 def _getPluginCommands(cls) -> defaultdict[str, list[str | PluginCommand]]:
309 """Get the commands offered by plugin packages.
311 Returns
312 -------
313 commands : `defaultdict` [`str`, `list` [`str`]]
314 The key is the command name. The value is a list of package(s) that
315 contains the command.
317 Notes
318 -----
319 Assumes that if entry points are defined, the plugin environment
320 variable will not be defined for that same package.
321 """
322 commands: defaultdict[str, list[str | PluginCommand]] = defaultdict(list)
323 for pluginName in cls.getPluginList():
324 try:
325 resources = defaultdict(list, yaml.safe_load(pluginName.read()))
326 except Exception as err:
327 log.warning("Error loading commands from %s, skipping. %s", pluginName, err)
328 continue
329 if "cmd" not in resources:
330 log.warning("No commands found in %s, skipping.", pluginName)
331 continue
332 pluginCommands = {cmd: [resources["cmd"]["import"]] for cmd in resources["cmd"]["commands"]}
333 cls._mergeCommandLists(commands, defaultdict(list, pluginCommands))
335 if hasattr(cls, "entryPoint"):
336 plugins = entry_points(group=cls.entryPoint)
337 for p in plugins:
338 try:
339 func = p.load()
340 except Exception as err:
341 log.warning("Could not import plugin from entry point %s, skipping.", p)
342 log.debug(
343 "Plugin import exception: %s\nTraceback:\n%s",
344 err,
345 "".join(traceback.format_tb(err.__traceback__)),
346 )
347 continue
348 func_name = get_full_type_name(func)
349 pluginCommands = {cmd.name: [PluginCommand(cmd, func_name)] for cmd in func()}
350 cls._mergeCommandLists(commands, defaultdict(list, pluginCommands))
352 return commands
354 @cache
355 def _getCommands(self) -> defaultdict[str, list[str | PluginCommand]]:
356 """Get the commands offered by daf_butler and plugin packages.
358 Returns
359 -------
360 commands : `defaultdict` [`str`, `list` [`str` | `PluginCommand` ]]
361 The key is the command name. The value is a list of package(s) that
362 contains the command.
363 """
364 return self._mergeCommandLists(self.getLocalCommands(), self._getPluginCommands())
366 @staticmethod
367 def _raiseIfDuplicateCommands(commands: defaultdict[str, list[str | PluginCommand]]) -> None:
368 """If any provided command is offered by more than one package raise an
369 exception.
371 Parameters
372 ----------
373 commands : `defaultdict` [`str`, `list` [`str` | `PLuginCommand` ]]
374 The key is the command name. The value is a list of package(s) that
375 contains the command.
377 Raises
378 ------
379 click.ClickException
380 Raised if a command is offered by more than one package, with an
381 error message to be displayed to the user.
382 """
383 msg = ""
384 for command, packages in commands.items():
385 if len(packages) > 1:
386 pkg_names: list[str] = []
387 for p in packages:
388 if not isinstance(p, str):
389 p = p.source
390 pkg_names.append(p)
391 msg += f"Command '{command}' exists in packages {', '.join(pkg_names)}. "
392 if msg:
393 raise click.ClickException(message=msg + "Duplicate commands are not supported, aborting.")
396class ButlerCLI(LoaderCLI):
397 """Specialized command loader implementing the ``butler`` command."""
399 localCmdPkg = "lsst.daf.butler.cli.cmd"
401 pluginEnvVar = "DAF_BUTLER_PLUGINS"
402 entryPoint = "butler.cli"
404 @classmethod
405 def _funcNameToCmdName(cls, functionName: str) -> str:
406 # Docstring inherited from base class.
408 # The "import" command name and "butler_import" function name are
409 # defined in cli/cmd/commands.py, and if those names are changed they
410 # must be changed here as well.
411 # It is expected that there will be very few butler command names that
412 # need to be changed because of e.g. conflicts with python keywords (as
413 # is done here and in _cmdNameToFuncName for the 'import' command). If
414 # this becomes a common need then some way of doing this should be
415 # invented that is better than hard coding the function names into
416 # these conversion functions. An extension of the 'cli/resources.yaml'
417 # file (as is currently used in obs_base) might be a good way to do it.
418 if functionName == "butler_import":
419 return "import"
420 return super()._funcNameToCmdName(functionName)
422 @classmethod
423 def _cmdNameToFuncName(cls, commandName: str) -> str:
424 # Docstring inherited from base class.
425 if commandName == "import":
426 return "butler_import"
427 return super()._cmdNameToFuncName(commandName)
430class UncachedButlerCLI(ButlerCLI):
431 """ButlerCLI that can be used where caching of the commands is disabled."""
433 def _getCommands(self) -> defaultdict[str, list[str | PluginCommand]]: # type: ignore[override]
434 """Get the commands offered by daf_butler and plugin packages.
436 Returns
437 -------
438 commands : `defaultdict` [`str`, `list` [`str` | `PluginCommand` ]]
439 The key is the command name. The value is a list of package(s) that
440 contains the command.
441 """
442 return self._mergeCommandLists(self.getLocalCommands(), self._getPluginCommands())
445@click.command(cls=ButlerCLI, context_settings=dict(help_option_names=["-h", "--help"]))
446@log_level_option()
447@long_log_option()
448@log_file_option()
449@log_tty_option()
450@log_label_option()
451@ClickProgressHandler.option
452def cli(log_level: str, long_log: bool, log_file: str, log_tty: bool, log_label: str, progress: bool) -> None:
453 """Butler command-line tools.
455 Log options apply to all subcommands.
456 """
457 pass
460def main() -> click.Command:
461 """Return main entry point for command-line."""
462 return cli()