20565403496

Committed 29 Dec 2025 05:11AM UTC coverage: 84.103% (-2.8%) from 86.921%

Build # 20565403496

Build Type

Pull #13567

github

Committed by

web-flow

Commit Message

Merge 4816837a5 into 2417384aa

Pull Request Pull Request #13567: Update ASF APIs

Run Details

67166 of 79862 relevant lines covered (84.1%)

0.84 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

88.31

/localstack-core/localstack/packages/api.py

import abc
import functools
import logging
import os
from collections import defaultdict
from collections.abc import Callable
from enum import Enum
from inspect import getmodule
from threading import Lock, RLock
from typing import Any, Generic, ParamSpec, TypeVar

from plux import Plugin, PluginManager, PluginSpec  # type: ignore

from localstack import config

LOG = logging.getLogger(__name__)


class PackageException(Exception):
    """Basic exception indicating that a package-specific exception occurred."""

    pass


class NoSuchVersionException(PackageException):
    """Exception indicating that a requested installer version is not available / supported."""

    def __init__(self, package: str | None = None, version: str | None = None):
        message = "Unable to find requested version"
        if package and version:
            message += f"Unable to find requested version '{version}' for package '{package}'"
        super().__init__(message)


class InstallTarget(Enum):
    """
    Different installation targets.
    Attention:
    - These targets are directly used in the LPM API and are therefore part of a public API!
    - The order of the entries in the enum define the default lookup order when looking for package installations.

    These targets refer to the directories in config#Directories.
    - VAR_LIBS: Used for packages installed at runtime. They are installed in a host-mounted volume.
                This directory / these installations persist across multiple containers.
    - STATIC_LIBS: Used for packages installed at build time. They are installed in a non-host-mounted volume.
                   This directory is re-created whenever a container is recreated.
    """

    VAR_LIBS = config.dirs.var_libs
    STATIC_LIBS = config.dirs.static_libs


class PackageInstaller(abc.ABC):
    """
    Base class for a specific installer.
    An instance of an installer manages the installation of a specific Package (in a specific version, if there are
    multiple versions).
    """

    def __init__(self, name: str, version: str, install_lock: Lock | None = None):
        """
        :param name: technical package name, f.e. "opensearch"
        :param version: version of the package to install
        :param install_lock: custom lock which should be used for this package installer instance for the
                             complete #install call. Defaults to a per-instance reentrant lock (RLock).
                             Package instances create one installer per version. Therefore, by default, the lock
                             ensures that package installations of the same package and version are mutually exclusive.
        """
        self.name = name
        self.version = version
        self.install_lock = install_lock or RLock()
        self._setup_for_target: dict[InstallTarget, bool] = defaultdict(lambda: False)

    def install(self, target: InstallTarget | None = None) -> None:
        """
        Performs the package installation.

        :param target: preferred installation target. Default is VAR_LIBS.
        :return: None
        :raises PackageException: if the installation fails
        """
        try:
            if not target:
                target = InstallTarget.VAR_LIBS
            # We have to acquire the lock before checking if the package is installed, as the is_installed check
            # is _only_ reliable if no other thread is currently actually installing
            with self.install_lock:
                # Skip the installation if it's already installed
                if not self.is_installed():
                    LOG.debug("Starting installation of %s %s...", self.name, self.version)
                    self._prepare_installation(target)
                    self._install(target)
                    self._post_process(target)
                    LOG.debug("Installation of %s %s finished.", self.name, self.version)
                else:
                    LOG.debug(
                        "Installation of %s %s skipped (already installed).",
                        self.name,
                        self.version,
                    )
                    if not self._setup_for_target[target]:
                        LOG.debug("Performing runtime setup for already installed package.")
                        self._setup_existing_installation(target)
        except PackageException as e:
            raise e
        except Exception as e:
            raise PackageException(f"Installation of {self.name} {self.version} failed.") from e

    def is_installed(self) -> bool:
        """
        Checks if the package is already installed.

        :return: True if the package is already installed (i.e. an installation is not necessary).
        """
        return self.get_installed_dir() is not None

    def get_installed_dir(self) -> str | None:
        """
        Returns the directory of an existing installation. The directory can differ based on the installation target
        and version.
        :return: str representation of the installation directory path or None if the package is not installed anywhere
        """
        for target in InstallTarget:
            directory = self._get_install_dir(target)
            if directory and os.path.exists(self._get_install_marker_path(directory)):
                return directory
        return None

    def _get_install_dir(self, target: InstallTarget) -> str:
        """
        Builds the installation directory for a specific target.
        :param target: to create the installation directory path for
        :return: str representation of the installation directory for the given target
        """
        return os.path.join(target.value, self.name, self.version)

    def _get_install_marker_path(self, install_dir: str) -> str:
        """
        Builds the path for a specific "marker" whose presence indicates that the package has been installed
        successfully in the given directory.

        :param install_dir: base path for the check (f.e. /var/lib/localstack/lib/dynamodblocal/latest/)
        :return: path which should be checked to indicate if the package has been installed successfully
                 (f.e. /var/lib/localstack/lib/dynamodblocal/latest/DynamoDBLocal.jar)
        """
        raise NotImplementedError()

    def _setup_existing_installation(self, target: InstallTarget) -> None:
        """
        Internal function to perform the setup for an existing installation, f.e. adding a path to an environment.
        This is only necessary for certain installers (like the PythonPackageInstaller).
        This function will _always_ be executed _exactly_ once within a Python session for a specific installer
        instance and target, if #install is called for the respective target.
        :param target: of the installation
        :return: None
        """
        pass

    def _prepare_installation(self, target: InstallTarget) -> None:
        """
        Internal function to prepare an installation, f.e. by downloading some data or installing an OS package repo.
        Can be implemented by specific installers.
        :param target: of the installation
        :return: None
        """
        pass

    def _install(self, target: InstallTarget) -> None:
        """
        Internal function to perform the actual installation.
        Must be implemented by specific installers.
        :param target: of the installation
        :return: None
        """
        raise NotImplementedError()

    def _post_process(self, target: InstallTarget) -> None:
        """
        Internal function to perform some post-processing, f.e. patching an installation or creating symlinks.
        :param target: of the installation
        :return: None
        """
        pass


# With Python 3.13 we should be able to set PackageInstaller as the default
# https://typing.python.org/en/latest/spec/generics.html#type-parameter-defaults
T = TypeVar("T", bound=PackageInstaller)


class Package(abc.ABC, Generic[T]):
    """
    A Package defines a specific kind of software, mostly used as backends or supporting system for service
    implementations.
    """

    def __init__(self, name: str, default_version: str):
        """
        :param name: Human readable name of the package, f.e. "PostgreSQL"
        :param default_version: Default version of the package which is used for installations if no version is defined
        """
        self.name = name
        self.default_version = default_version

    def get_installed_dir(self, version: str | None = None) -> str | None:
        """
        Finds a directory where the package (in the specific version) is installed.
        :param version: of the package to look for. If None, the default version of the package is used.
        :return: str representation of the path to the existing installation directory or None if the package in this
                 version is not yet installed.
        """
        return self.get_installer(version).get_installed_dir()

    def install(self, version: str | None = None, target: InstallTarget | None = None) -> None:
        """
        Installs the package in the given version in the preferred target location.
        :param version: version of the package to install. If None, the default version of the package will be used.
        :param target: preferred installation target. If None, the var_libs directory is used.
        :raises NoSuchVersionException: If the given version is not supported.
        """
        self.get_installer(version).install(target)

    @functools.lru_cache
    def get_installer(self, version: str | None = None) -> T:
        """
        Returns the installer instance for a specific version of the package.

        It is important that this be LRU cached. Installers have a mutex lock to prevent races, and it is necessary
        that this method returns the same installer instance for a given version.

        :param version: version of the package to install. If None, the default version of the package will be used.
        :return: PackageInstaller instance for the given version.
        :raises NoSuchVersionException: If the given version is not supported.
        """
        if not version:
            return self.get_installer(self.default_version)
        if version not in self.get_versions():
            raise NoSuchVersionException(package=self.name, version=version)
        return self._get_installer(version)

    def get_versions(self) -> list[str]:
        """
        :return: List of all versions available for this package.
        """
        raise NotImplementedError()

    def _get_installer(self, version: str) -> T:
        """
        Internal lookup function which needs to be implemented by specific packages.
        It creates PackageInstaller instances for the specific version.

        :param version: to find the installer for
        :return: PackageInstaller instance responsible for installing the given version of the package.
        """
        raise NotImplementedError()

    def __str__(self) -> str:
        return self.name


class MultiPackageInstaller(PackageInstaller):
    """
    PackageInstaller implementation which composes of multiple package installers.
    """

    def __init__(self, name: str, version: str, package_installer: list[PackageInstaller]):
        """
        :param name: of the (multi-)package installer
        :param version: of this (multi-)package installer
        :param package_installer: List of installers this multi-package installer consists of
        """
        super().__init__(name=name, version=version)

        assert isinstance(package_installer, list)
        assert len(package_installer) > 0
        self.package_installer = package_installer

    def install(self, target: InstallTarget | None = None) -> None:
        """
        Installs the different packages this installer is composed of.

        :param target: which defines where to install the packages.
        :return: None
        """
        for package_installer in self.package_installer:
            package_installer.install(target=target)

    def get_installed_dir(self) -> str | None:
        # By default, use the installed-dir of the first package
        return self.package_installer[0].get_installed_dir()

    def _install(self, target: InstallTarget) -> None:
        # This package installer actually only calls other installers, we pass here
        pass

    def _get_install_dir(self, target: InstallTarget) -> str:
        # By default, use the install-dir of the first package
        return self.package_installer[0]._get_install_dir(target)

    def _get_install_marker_path(self, install_dir: str) -> str:
        # By default, use the install-marker-path of the first package
        return self.package_installer[0]._get_install_marker_path(install_dir)


PLUGIN_NAMESPACE = "localstack.packages"


class PackagesPlugin(Plugin):  # type: ignore[misc]
    """
    Plugin implementation for Package plugins.
    A package plugin exposes a specific package instance.
    """

    api: str
    name: str

    def __init__(
        self,
        name: str,
        scope: str,
        get_package: Callable[[], Package[PackageInstaller] | list[Package[PackageInstaller]]],
        should_load: Callable[[], bool] | None = None,
    ) -> None:
        super().__init__()
        self.name = name
        self.scope = scope
        self._get_package = get_package
        self._should_load = should_load

    def should_load(self) -> bool:
        if self._should_load:
            return self._should_load()
        return True

    def get_package(self) -> Package[PackageInstaller]:
        """
        :return: returns the package instance of this package plugin
        """
        return self._get_package()  # type: ignore[return-value]


class NoSuchPackageException(PackageException):
    """Exception raised by the PackagesPluginManager to indicate that a package / version is not available."""

    pass


class PackagesPluginManager(PluginManager[PackagesPlugin]):  # type: ignore[misc]
    """PluginManager which simplifies the loading / access of PackagesPlugins and their exposed package instances."""

    def __init__(self) -> None:
        super().__init__(PLUGIN_NAMESPACE)

    def get_all_packages(self) -> list[tuple[str, str, Package[PackageInstaller]]]:
        return sorted(
            [(plugin.name, plugin.scope, plugin.get_package()) for plugin in self.load_all()]
        )

    def get_packages(
        self, package_names: list[str], version: str | None = None
    ) -> list[Package[PackageInstaller]]:
        # Plugin names are unique, but there could be multiple packages with the same name in different scopes
        plugin_specs_per_name = defaultdict(list)
        # Plugin names have the format "<package-name>/<scope>", build a dict of specs per package name for the lookup
        for plugin_spec in self.list_plugin_specs():
            (package_name, _, _) = plugin_spec.name.rpartition("/")
            plugin_specs_per_name[package_name].append(plugin_spec)

        package_instances: list[Package[PackageInstaller]] = []
        for package_name in package_names:
            plugin_specs = plugin_specs_per_name.get(package_name)
            if not plugin_specs:
                raise NoSuchPackageException(
                    f"unable to locate installer for package {package_name}"
                )
            for plugin_spec in plugin_specs:
                package_instance = self.load(plugin_spec.name).get_package()
                package_instances.append(package_instance)
                if version and version not in package_instance.get_versions():
                    raise NoSuchPackageException(
                        f"unable to locate installer for package {package_name} and version {version}"
                    )

        return package_instances


P = ParamSpec("P")
T2 = TypeVar("T2")


def package(
    name: str | None = None,
    scope: str = "community",
    should_load: Callable[[], bool] | None = None,
) -> Callable[[Callable[[], Package[Any] | list[Package[Any]]]], PluginSpec]:
    """
    Decorator for marking methods that create Package instances as a PackagePlugin.
    Methods marked with this decorator are discoverable as a PluginSpec within the namespace "localstack.packages",
    with the name "<name>:<scope>". If api is not explicitly specified, then the parent module name is used as
    service name.
    """

    def wrapper(fn: Callable[[], Package[Any] | list[Package[Any]]]) -> PluginSpec:
        _name = name or getmodule(fn).__name__.split(".")[-2]  # type: ignore[union-attr]

        @functools.wraps(fn)
        def factory() -> PackagesPlugin:
            return PackagesPlugin(name=_name, scope=scope, get_package=fn, should_load=should_load)

        return PluginSpec(PLUGIN_NAMESPACE, f"{_name}/{scope}", factory=factory)

    return wrapper


# TODO remove (only used for migrating to new #package decorator)
packages = package

1	import abc	1✔
2	import functools	1✔
3	import logging	1✔
4	import os	1✔
5	from collections import defaultdict	1✔
6	from collections.abc import Callable	1✔
7	from enum import Enum	1✔
8	from inspect import getmodule	1✔
9	from threading import Lock, RLock	1✔
10	from typing import Any, Generic, ParamSpec, TypeVar	1✔
11
12	from plux import Plugin, PluginManager, PluginSpec # type: ignore	1✔
13
14	from localstack import config	1✔
15
16	LOG = logging.getLogger(__name__)	1✔
17
18
19	class PackageException(Exception):	1✔
20	"""Basic exception indicating that a package-specific exception occurred."""
21
22	pass	1✔
23
24
25	class NoSuchVersionException(PackageException):	1✔
26	"""Exception indicating that a requested installer version is not available / supported."""
27
28	def __init__(self, package: str \| None = None, version: str \| None = None):	1✔
29	message = "Unable to find requested version"	×
30	if package and version:	×
31	message += f"Unable to find requested version '{version}' for package '{package}'"	×
32	super().__init__(message)	×
33
34
35	class InstallTarget(Enum):	1✔
36	"""
37	Different installation targets.
38	Attention:
39	- These targets are directly used in the LPM API and are therefore part of a public API!
40	- The order of the entries in the enum define the default lookup order when looking for package installations.
41
42	These targets refer to the directories in config#Directories.
43	- VAR_LIBS: Used for packages installed at runtime. They are installed in a host-mounted volume.
44	This directory / these installations persist across multiple containers.
45	- STATIC_LIBS: Used for packages installed at build time. They are installed in a non-host-mounted volume.
46	This directory is re-created whenever a container is recreated.
47	"""
48
49	VAR_LIBS = config.dirs.var_libs	1✔
50	STATIC_LIBS = config.dirs.static_libs	1✔
51
52
53	class PackageInstaller(abc.ABC):	1✔
54	"""
55	Base class for a specific installer.
56	An instance of an installer manages the installation of a specific Package (in a specific version, if there are
57	multiple versions).
58	"""
59
60	def __init__(self, name: str, version: str, install_lock: Lock \| None = None):	1✔
61	"""
62	:param name: technical package name, f.e. "opensearch"
63	:param version: version of the package to install
64	:param install_lock: custom lock which should be used for this package installer instance for the
65	complete #install call. Defaults to a per-instance reentrant lock (RLock).
66	Package instances create one installer per version. Therefore, by default, the lock
67	ensures that package installations of the same package and version are mutually exclusive.
68	"""
69	self.name = name	1✔
70	self.version = version	1✔
71	self.install_lock = install_lock or RLock()	1✔
72	self._setup_for_target: dict[InstallTarget, bool] = defaultdict(lambda: False)	1✔
73
74	def install(self, target: InstallTarget \| None = None) -> None:	1✔
75	"""
76	Performs the package installation.
77
78	:param target: preferred installation target. Default is VAR_LIBS.
79	:return: None
80	:raises PackageException: if the installation fails
81	"""
82	try:	1✔
83	if not target:	1✔
84	target = InstallTarget.VAR_LIBS	1✔
85	# We have to acquire the lock before checking if the package is installed, as the is_installed check
86	# is _only_ reliable if no other thread is currently actually installing
87	with self.install_lock:	1✔
88	# Skip the installation if it's already installed
89	if not self.is_installed():	1✔
90	LOG.debug("Starting installation of %s %s...", self.name, self.version)	1✔
91	self._prepare_installation(target)	1✔
92	self._install(target)	1✔
93	self._post_process(target)	1✔
94	LOG.debug("Installation of %s %s finished.", self.name, self.version)	1✔
95	else:
96	LOG.debug(	1✔
97	"Installation of %s %s skipped (already installed).",
98	self.name,
99	self.version,
100	)
101	if not self._setup_for_target[target]:	1✔
102	LOG.debug("Performing runtime setup for already installed package.")	1✔
103	self._setup_existing_installation(target)	1✔
104	except PackageException as e:	1✔
105	raise e	1✔
106	except Exception as e:	×
107	raise PackageException(f"Installation of {self.name} {self.version} failed.") from e	×
108
109	def is_installed(self) -> bool:	1✔
110	"""
111	Checks if the package is already installed.
112
113	:return: True if the package is already installed (i.e. an installation is not necessary).
114	"""
115	return self.get_installed_dir() is not None	1✔
116
117	def get_installed_dir(self) -> str \| None:	1✔
118	"""
119	Returns the directory of an existing installation. The directory can differ based on the installation target
120	and version.
121	:return: str representation of the installation directory path or None if the package is not installed anywhere
122	"""
123	for target in InstallTarget:	1✔
124	directory = self._get_install_dir(target)	1✔
125	if directory and os.path.exists(self._get_install_marker_path(directory)):	1✔
126	return directory	1✔
127	return None	1✔
128
129	def _get_install_dir(self, target: InstallTarget) -> str:	1✔
130	"""
131	Builds the installation directory for a specific target.
132	:param target: to create the installation directory path for
133	:return: str representation of the installation directory for the given target
134	"""
135	return os.path.join(target.value, self.name, self.version)	1✔
136
137	def _get_install_marker_path(self, install_dir: str) -> str:	1✔
138	"""
139	Builds the path for a specific "marker" whose presence indicates that the package has been installed
140	successfully in the given directory.
141
142	:param install_dir: base path for the check (f.e. /var/lib/localstack/lib/dynamodblocal/latest/)
143	:return: path which should be checked to indicate if the package has been installed successfully
144	(f.e. /var/lib/localstack/lib/dynamodblocal/latest/DynamoDBLocal.jar)
145	"""
146	raise NotImplementedError()
147
148	def _setup_existing_installation(self, target: InstallTarget) -> None:	1✔
149	"""
150	Internal function to perform the setup for an existing installation, f.e. adding a path to an environment.
151	This is only necessary for certain installers (like the PythonPackageInstaller).
152	This function will _always_ be executed _exactly_ once within a Python session for a specific installer
153	instance and target, if #install is called for the respective target.
154	:param target: of the installation
155	:return: None
156	"""
157	pass	1✔
158
159	def _prepare_installation(self, target: InstallTarget) -> None:	1✔
160	"""
161	Internal function to prepare an installation, f.e. by downloading some data or installing an OS package repo.
162	Can be implemented by specific installers.
163	:param target: of the installation
164	:return: None
165	"""
166	pass	1✔
167
168	def _install(self, target: InstallTarget) -> None:	1✔
169	"""
170	Internal function to perform the actual installation.
171	Must be implemented by specific installers.
172	:param target: of the installation
173	:return: None
174	"""
175	raise NotImplementedError()
176
177	def _post_process(self, target: InstallTarget) -> None:	1✔
178	"""
179	Internal function to perform some post-processing, f.e. patching an installation or creating symlinks.
180	:param target: of the installation
181	:return: None
182	"""
183	pass	1✔
184
185
186	# With Python 3.13 we should be able to set PackageInstaller as the default
187	# https://typing.python.org/en/latest/spec/generics.html#type-parameter-defaults
188	T = TypeVar("T", bound=PackageInstaller)	1✔
189
190
191	class Package(abc.ABC, Generic[T]):	1✔
192	"""
193	A Package defines a specific kind of software, mostly used as backends or supporting system for service
194	implementations.
195	"""
196
197	def __init__(self, name: str, default_version: str):	1✔
198	"""
199	:param name: Human readable name of the package, f.e. "PostgreSQL"
200	:param default_version: Default version of the package which is used for installations if no version is defined
201	"""
202	self.name = name	1✔
203	self.default_version = default_version	1✔
204
205	def get_installed_dir(self, version: str \| None = None) -> str \| None:	1✔
206	"""
207	Finds a directory where the package (in the specific version) is installed.
208	:param version: of the package to look for. If None, the default version of the package is used.
209	:return: str representation of the path to the existing installation directory or None if the package in this
210	version is not yet installed.
211	"""
212	return self.get_installer(version).get_installed_dir()	1✔
213
214	def install(self, version: str \| None = None, target: InstallTarget \| None = None) -> None:	1✔
215	"""
216	Installs the package in the given version in the preferred target location.
217	:param version: version of the package to install. If None, the default version of the package will be used.
218	:param target: preferred installation target. If None, the var_libs directory is used.
219	:raises NoSuchVersionException: If the given version is not supported.
220	"""
221	self.get_installer(version).install(target)	1✔
222
223	@functools.lru_cache	1✔
224	def get_installer(self, version: str \| None = None) -> T:	1✔
225	"""
226	Returns the installer instance for a specific version of the package.
227
228	It is important that this be LRU cached. Installers have a mutex lock to prevent races, and it is necessary
229	that this method returns the same installer instance for a given version.
230
231	:param version: version of the package to install. If None, the default version of the package will be used.
232	:return: PackageInstaller instance for the given version.
233	:raises NoSuchVersionException: If the given version is not supported.
234	"""
235	if not version:	1✔
236	return self.get_installer(self.default_version)	1✔
237	if version not in self.get_versions():	1✔
238	raise NoSuchVersionException(package=self.name, version=version)	×
239	return self._get_installer(version)	1✔
240
241	def get_versions(self) -> list[str]:	1✔
242	"""
243	:return: List of all versions available for this package.
244	"""
245	raise NotImplementedError()
246
247	def _get_installer(self, version: str) -> T:	1✔
248	"""
249	Internal lookup function which needs to be implemented by specific packages.
250	It creates PackageInstaller instances for the specific version.
251
252	:param version: to find the installer for
253	:return: PackageInstaller instance responsible for installing the given version of the package.
254	"""
255	raise NotImplementedError()
256
257	def __str__(self) -> str:	1✔
258	return self.name	1✔
259
260
261	class MultiPackageInstaller(PackageInstaller):	1✔
262	"""
263	PackageInstaller implementation which composes of multiple package installers.
264	"""
265
266	def __init__(self, name: str, version: str, package_installer: list[PackageInstaller]):	1✔
267	"""
268	:param name: of the (multi-)package installer
269	:param version: of this (multi-)package installer
270	:param package_installer: List of installers this multi-package installer consists of
271	"""
272	super().__init__(name=name, version=version)	×
273
274	assert isinstance(package_installer, list)	×
275	assert len(package_installer) > 0	×
276	self.package_installer = package_installer	×
277
278	def install(self, target: InstallTarget \| None = None) -> None:	1✔
279	"""
280	Installs the different packages this installer is composed of.
281
282	:param target: which defines where to install the packages.
283	:return: None
284	"""
285	for package_installer in self.package_installer:	×
286	package_installer.install(target=target)	×
287
288	def get_installed_dir(self) -> str \| None:	1✔
289	# By default, use the installed-dir of the first package
290	return self.package_installer[0].get_installed_dir()	×
291
292	def _install(self, target: InstallTarget) -> None:	1✔
293	# This package installer actually only calls other installers, we pass here
294	pass	×
295
296	def _get_install_dir(self, target: InstallTarget) -> str:	1✔
297	# By default, use the install-dir of the first package
298	return self.package_installer[0]._get_install_dir(target)	×
299
300	def _get_install_marker_path(self, install_dir: str) -> str:	1✔
301	# By default, use the install-marker-path of the first package
302	return self.package_installer[0]._get_install_marker_path(install_dir)	×
303
304
305	PLUGIN_NAMESPACE = "localstack.packages"	1✔
306
307
308	class PackagesPlugin(Plugin): # type: ignore[misc]	1✔
309	"""
310	Plugin implementation for Package plugins.
311	A package plugin exposes a specific package instance.
312	"""
313
314	api: str	1✔
315	name: str	1✔
316
317	def __init__(	1✔
318	self,
319	name: str,
320	scope: str,
321	get_package: Callable[[], Package[PackageInstaller] \| list[Package[PackageInstaller]]],
322	should_load: Callable[[], bool] \| None = None,
323	) -> None:
324	super().__init__()	1✔
325	self.name = name	1✔
326	self.scope = scope	1✔
327	self._get_package = get_package	1✔
328	self._should_load = should_load	1✔
329
330	def should_load(self) -> bool:	1✔
331	if self._should_load:	1✔
332	return self._should_load()	×
333	return True	1✔
334
335	def get_package(self) -> Package[PackageInstaller]:	1✔
336	"""
337	:return: returns the package instance of this package plugin
338	"""
339	return self._get_package() # type: ignore[return-value]	1✔
340
341
342	class NoSuchPackageException(PackageException):	1✔
343	"""Exception raised by the PackagesPluginManager to indicate that a package / version is not available."""
344
345	pass	1✔
346
347
348	class PackagesPluginManager(PluginManager[PackagesPlugin]): # type: ignore[misc]	1✔
349	"""PluginManager which simplifies the loading / access of PackagesPlugins and their exposed package instances."""
350
351	def __init__(self) -> None:	1✔
352	super().__init__(PLUGIN_NAMESPACE)	1✔
353
354	def get_all_packages(self) -> list[tuple[str, str, Package[PackageInstaller]]]:	1✔
355	return sorted(	1✔
356	[(plugin.name, plugin.scope, plugin.get_package()) for plugin in self.load_all()]
357	)
358
359	def get_packages(	1✔
360	self, package_names: list[str], version: str \| None = None
361	) -> list[Package[PackageInstaller]]:
362	# Plugin names are unique, but there could be multiple packages with the same name in different scopes
363	plugin_specs_per_name = defaultdict(list)	1✔
364	# Plugin names have the format "<package-name>/<scope>", build a dict of specs per package name for the lookup
365	for plugin_spec in self.list_plugin_specs():	1✔
366	(package_name, _, _) = plugin_spec.name.rpartition("/")	1✔
367	plugin_specs_per_name[package_name].append(plugin_spec)	1✔
368
369	package_instances: list[Package[PackageInstaller]] = []	1✔
370	for package_name in package_names:	1✔
371	plugin_specs = plugin_specs_per_name.get(package_name)	1✔
372	if not plugin_specs:	1✔
373	raise NoSuchPackageException(	1✔
374	f"unable to locate installer for package {package_name}"
375	)
376	for plugin_spec in plugin_specs:	1✔
377	package_instance = self.load(plugin_spec.name).get_package()	1✔
378	package_instances.append(package_instance)	1✔
379	if version and version not in package_instance.get_versions():	1✔
380	raise NoSuchPackageException(	1✔
381	f"unable to locate installer for package {package_name} and version {version}"
382	)
383
384	return package_instances	1✔
385
386
387	P = ParamSpec("P")	1✔
388	T2 = TypeVar("T2")	1✔
389
390
391	def package(	1✔
392	name: str \| None = None,
393	scope: str = "community",
394	should_load: Callable[[], bool] \| None = None,
395	) -> Callable[[Callable[[], Package[Any] \| list[Package[Any]]]], PluginSpec]:
396	"""
397	Decorator for marking methods that create Package instances as a PackagePlugin.
398	Methods marked with this decorator are discoverable as a PluginSpec within the namespace "localstack.packages",
399	with the name "<name>:<scope>". If api is not explicitly specified, then the parent module name is used as
400	service name.
401	"""
402
403	def wrapper(fn: Callable[[], Package[Any] \| list[Package[Any]]]) -> PluginSpec:	1✔
404	_name = name or getmodule(fn).__name__.split(".")[-2] # type: ignore[union-attr]	1✔
405
406	@functools.wraps(fn)	1✔
407	def factory() -> PackagesPlugin:	1✔
408	return PackagesPlugin(name=_name, scope=scope, get_package=fn, should_load=should_load)	1✔
409
410	return PluginSpec(PLUGIN_NAMESPACE, f"{_name}/{scope}", factory=factory)	1✔
411
412	return wrapper	1✔
413
414
415	# TODO remove (only used for migrating to new #package decorator)
416	packages = package	1✔

localstack / localstack / 20565403496

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous