6eb2aab0-9781-40f0-8cc1-9e7c437858cb

Committed 22 Apr 2025 04:28PM UTC coverage: 86.272% (+0.007%) from 86.265%

Build # 6eb2aab0-9781-40f0-8cc1-9e7c437858cb

Build Type

push

circleci

Committed by

web-flow

Commit Message

APIGW: validate REST API custom id tag (#12539)

Coverage Stats

7 of 7 new or added lines in 1 file covered. (100.0%)

82 existing lines in 10 files now uncovered.

63897 of 74065 relevant lines covered (86.27%)

0.86 hits per line

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

88.24

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

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

from plux import Plugin, PluginManager, PluginSpec  # type: ignore[import-untyped]

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

localstack / localstack / 6eb2aab0-9781-40f0-8cc1-9e7c437858cb

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