Spaces:
Sleeping
Sleeping
File size: 6,852 Bytes
53dbcc1 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 | """Lazy-loadable command specification for deferred imports."""
import importlib
from itertools import chain
from typing import TYPE_CHECKING, Any
from attrs import Factory, define, field
if TYPE_CHECKING:
from cyclopts.core import App
from cyclopts.group import Group
@define
class CommandSpec:
"""Specification for a command that will be lazily loaded on first access.
This allows registering commands via import path strings (e.g., "myapp.commands:create")
without importing them until they're actually used, improving CLI startup time.
Parameters
----------
import_path : str
Import path in the format "module.path:attribute_name".
The attribute should be either a function or an App instance.
name : str | tuple[str, ...] | None
CLI command name. If None, will be derived from the attribute name via name_transform.
For function imports: used as the name of the wrapper App.
For App imports: must match the App's internal name, or ValueError is raised at resolution.
app_kwargs : dict
Keyword arguments to pass to App() if wrapping a function.
Raises ValueError if used with App imports (Apps should be configured in their own definition).
Examples
--------
>>> from cyclopts import App
>>> app = App()
>>> # Lazy load - doesn't import myapp.commands until "create" is executed
>>> app.command("myapp.commands:create_user", name="create")
>>> app()
"""
import_path: str
name: str | tuple[str, ...] | None = None
app_kwargs: dict[str, Any] = Factory(dict)
help: str | None = None
sort_key: Any = None
group: "Group | str | tuple[Group | str, ...] | None" = None
_show: bool | None = field(default=None, alias="show")
@property
def show(self) -> bool:
if self._show is None:
return True
return self._show
_resolved: "App | None" = field(init=False, default=None, repr=False)
def resolve(self, parent_app: "App") -> "App":
"""Import and resolve the command on first access.
Parameters
----------
parent_app : App
Parent app to inherit defaults from (help_flags, version_flags, groups).
Required to match the behavior of direct command registration.
Returns
-------
App
The resolved App instance, either imported directly or wrapping a function.
Raises
------
ValueError
If import_path is not in the correct format "module.path:attribute_name".
ImportError
If the module cannot be imported.
AttributeError
If the attribute doesn't exist in the module.
"""
if self._resolved is not None:
return self._resolved
# Parse import path
module_path, _, attr_name = self.import_path.rpartition(":")
if not module_path or not attr_name:
raise ValueError(
f"Invalid import path: {self.import_path!r}. Expected format: 'module.path:attribute_name'"
)
# Import the module and get the attribute
try:
module = importlib.import_module(module_path)
except ImportError as e:
raise ImportError(f"Cannot import module {module_path!r} from {self.import_path!r}") from e
try:
target = getattr(module, attr_name)
except AttributeError as e:
raise AttributeError(
f"Module {module_path!r} has no attribute {attr_name!r} (from import path {self.import_path!r})"
) from e
# Wrap in App if needed
from cyclopts.core import App
if isinstance(target, App):
# Validate that no kwargs were provided for App imports
if self.app_kwargs:
raise ValueError(
f"Cannot apply configuration to imported App. "
f"Import path {self.import_path!r} resolves to an App, "
f"but kwargs were specified: {self.app_kwargs!r}. "
f"Configure the App in its definition instead."
)
# Validate that the App's name matches the expected CLI command name
# The name used for CLI registration is stored in self.name
if self.name is not None and target.name[0] != self.name:
raise ValueError(
f"Imported App name mismatch. "
f"Import path {self.import_path!r} resolves to an App with name={target.name[0]!r}, "
f"but it was registered with CLI command name={self.name!r}. "
f"Either use app.command('{self.import_path}', name='{target.name[0]}') "
f"or change the App's name to match."
)
# Copy parent groups if not set (matches direct App registration behavior)
from cyclopts.core import _apply_parent_defaults_to_app
_apply_parent_defaults_to_app(target, parent_app)
self._resolved = target
else:
# It's a function - wrap it in an App with parent defaults
# Match the behavior of direct function registration
app_kwargs = dict(self.app_kwargs) # Copy to avoid mutating
from cyclopts.core import _apply_parent_groups_to_kwargs
app_kwargs.setdefault("help_flags", parent_app.help_flags)
app_kwargs.setdefault("version_flags", parent_app.version_flags)
if "version" not in app_kwargs and parent_app.version is not None:
app_kwargs["version"] = parent_app.version
_apply_parent_groups_to_kwargs(app_kwargs, parent_app)
self._resolved = App(name=self.name, **app_kwargs)
self._resolved.default(target)
# Apply registration-time overrides to the resolved App
if self.help is not None:
self._resolved.help = self.help
if self.sort_key is not None:
self._resolved.sort_key = self.sort_key
if self.group is not None:
self._resolved.group = self.group
if self._show is not None:
self._resolved.show = self._show
if self._resolved._name_transform is None:
self._resolved.name_transform = parent_app.name_transform
# Hide help and version flags from subapp help output
# This matches the behavior of direct App/function registration in core.py
for flag in chain(self._resolved.help_flags, self._resolved.version_flags):
self._resolved[flag].show = False
return self._resolved
@property
def is_resolved(self) -> bool:
"""Check if this command has been imported and resolved yet."""
return self._resolved is not None
|