platypush/platypush
platypush
/
platypush
2
2
Fork 4
Code Issues 46 Pull Requests Packages Projects 4 Releases Wiki Activity

Full rewrite of `platypush.utils.manifest`. 

The new version encapsulates all the utility functions into three
classes - `Manifest`, `Manifests` and `Dependencies`.
This commit is contained in:
Fabio Manganiello 2023-08-19 13:28:40 +02:00
parent a8255f3621
commit dd3a701a2e
Signed by: blacklight
GPG Key ID: D90FBA7F76362774
2 changed files with 281 additions and 145 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

425
platypush/utils/manifest.py
View File

@ -1,99 +1,328 @@
from dataclasses import dataclass, field
import enum
import importlib
import inspect
import json
import logging
import os
import pathlib
import shutil
import sys
from typing import (
Dict,
Generator,
List,
Optional,
Iterable,
Mapping,
Set,
Type,
Union,
)
import yaml
from abc import ABC, abstractmethod
from typing import Optional, Iterable, Mapping, Callable, Type
from platypush.message.event import Event
supported_package_managers = {
'pacman': 'pacman -S',
'apt': 'apt-get install',
'apk': ['apk', 'add', '--no-cache', '--no-progress'],
'apt': ['apt', 'install', '-y', '-q'],
'pacman': ['pacman', '-S', '--noconfirm', '--noprogressbar'],
}
_available_package_manager = None
logger = logging.getLogger(__name__)
class ManifestType(enum.Enum):
"""
Manifest types.
"""
PLUGIN = 'plugin'
BACKEND = 'backend'
class Manifest(ABC):
@dataclass
class Dependencies:
"""
Dependencies for a plugin/backend.
"""
before: List[str] = field(default_factory=list)
""" Commands to execute before the component is installed. """
packages: Set[str] = field(default_factory=set)
""" System packages required by the component. """
pip: Set[str] = field(default_factory=set)
""" pip dependencies. """
after: List[str] = field(default_factory=list)
""" Commands to execute after the component is installed. """
@classmethod
def from_config(
cls, conf_file: Optional[str] = None, pkg_manager: Optional[str] = None
) -> "Dependencies":
"""
Parse the required dependencies from a configuration file.
"""
deps = cls()
for manifest in Manifests.by_config(conf_file, pkg_manager=pkg_manager):
deps.before += manifest.install.before
deps.pip.update(manifest.install.pip)
deps.packages.update(manifest.install.packages)
deps.after += manifest.install.after
return deps
def to_pkg_install_commands(
self, pkg_manager: Optional[str] = None, skip_sudo: bool = False
) -> Generator[str, None, None]:
"""
Generates the package manager commands required to install the given
dependencies on the system.
:param pkg_manager: Force package manager to use (default: looks for
the one available on the system).
:param skip_sudo: Skip sudo when installing packages (default: it will
look if the current user is root and use sudo otherwise).
"""
wants_sudo = not skip_sudo and os.getuid() != 0
pkg_manager = pkg_manager or get_available_package_manager()
if self.packages and pkg_manager:
yield ' '.join(
[
*(['sudo'] if wants_sudo else []),
*supported_package_managers[pkg_manager],
*sorted(self.packages),
]
)
def to_pip_install_commands(self) -> Generator[str, None, None]:
"""
Generates the pip commands required to install the given dependencies on
the system.
"""
# Recent versions want an explicit --break-system-packages option when
# installing packages via pip outside of a virtual environment
wants_break_system_packages = (
sys.version_info > (3, 10)
and sys.prefix == sys.base_prefix # We're not in a venv
)
if self.pip:
yield (
'pip install -U --no-input --no-cache-dir '
+ ('--break-system-packages ' if wants_break_system_packages else '')
+ ' '.join(sorted(self.pip))
)
def to_install_commands(
self, pkg_manager: Optional[str] = None, skip_sudo: bool = False
) -> Generator[str, None, None]:
"""
Generates the commands required to install the given dependencies on
this system.
:param pkg_manager: Force package manager to use (default: looks for
the one available on the system).
:param skip_sudo: Skip sudo when installing packages (default: it will
look if the current user is root and use sudo otherwise).
"""
for cmd in self.before:
yield cmd
for cmd in self.to_pkg_install_commands(
pkg_manager=pkg_manager, skip_sudo=skip_sudo
):
yield cmd
for cmd in self.to_pip_install_commands():
yield cmd
for cmd in self.after:
yield cmd
class Manifest:
"""
Base class for plugin/backend manifests.
"""
def __init__(self, package: str, description: Optional[str] = None,
install: Optional[Iterable[str]] = None, events: Optional[Mapping] = None, **_):
def __init__(
self,
package: str,
description: Optional[str] = None,
install: Optional[Dict[str, Iterable[str]]] = None,
events: Optional[Mapping] = None,
pkg_manager: Optional[str] = None,
**_,
):
self._pkg_manager = pkg_manager or get_available_package_manager()
self.description = description
self.install = install or {}
self.events = events or {}
self.logger = logging.getLogger(__name__)
self.install = self._init_deps(install or {})
self.events = self._init_events(events or {})
self.package = package
self.component_name = '.'.join(package.split('.')[2:])
self.component = None
@classmethod
@property
@abstractmethod
def component_getter(self) -> Callable[[str], object]:
raise NotImplementedError
def _init_deps(self, install: Mapping[str, Iterable[str]]) -> Dependencies:
deps = Dependencies()
for key, items in install.items():
if key == 'pip':
deps.pip.update(items)
elif key == 'before':
deps.before += items
elif key == 'after':
deps.after += items
elif key == self._pkg_manager:
deps.packages.update(items)
return deps
@staticmethod
def _init_events(
events: Union[Iterable[str], Mapping[str, Optional[str]]]
) -> Dict[Type[Event], str]:
evt_dict = events if isinstance(events, Mapping) else {e: None for e in events}
ret = {}
for evt_name, doc in evt_dict.items():
evt_module_name, evt_class_name = evt_name.rsplit('.', 1)
try:
evt_module = importlib.import_module(evt_module_name)
evt_class = getattr(evt_module, evt_class_name)
except Exception as e:
raise AssertionError(f'Could not load event {evt_name}: {e}') from e
ret[evt_class] = doc or evt_class.__doc__
return ret
@classmethod
def from_file(cls, filename: str) -> "Manifest":
def from_file(cls, filename: str, pkg_manager: Optional[str] = None) -> "Manifest":
"""
Parse a manifest filename into a ``Manifest`` class.
"""
with open(str(filename), 'r') as f:
manifest = yaml.safe_load(f).get('manifest', {})
assert 'type' in manifest, f'Manifest file {filename} has no type field'
comp_type = ManifestType(manifest.pop('type'))
manifest_class = _manifest_class_by_type[comp_type]
return manifest_class(**manifest)
@classmethod
def from_class(cls, clazz) -> "Manifest":
return cls.from_file(os.path.dirname(inspect.getfile(clazz)))
@classmethod
def from_component(cls, comp) -> "Manifest":
return cls.from_class(comp.__class__)
def get_component(self):
try:
self.component = self.component_getter(self.component_name)
except Exception as e:
self.logger.warning(f'Could not load {self.component_name}: {e}')
return self.component
return manifest_class(**manifest, pkg_manager=pkg_manager)
def __repr__(self):
return json.dumps({
'description': self.description,
'install': self.install,
'events': self.events,
'type': _manifest_type_by_class[self.__class__].value,
'package': self.package,
'component_name': self.component_name,
})
"""
:return: A JSON serialized representation of the manifest.
"""
return json.dumps(
{
'description': self.description,
'install': self.install,
'events': {
'.'.join([evt_type.__module__, evt_type.__name__]): doc
for evt_type, doc in self.events.items()
},
'type': _manifest_type_by_class[self.__class__].value,
'package': self.package,
'component_name': self.component_name,
}
)
# pylint: disable=too-few-public-methods
class PluginManifest(Manifest):
@classmethod
@property
def component_getter(self):
from platypush.context import get_plugin
return get_plugin
"""
Plugin manifest.
"""
# pylint: disable=too-few-public-methods
class BackendManifest(Manifest):
@classmethod
@property
def component_getter(self):
from platypush.context import get_backend
return get_backend
"""
Backend manifest.
"""
class Manifests:
"""
General-purpose manifests utilities.
"""
@staticmethod
def by_base_class(
base_class: Type, pkg_manager: Optional[str] = None
) -> Generator[Manifest, None, None]:
"""
Get all the manifest files declared under the base path of a given class
and parse them into :class:`Manifest` objects.
"""
for mf in pathlib.Path(os.path.dirname(inspect.getfile(base_class))).rglob(
'manifest.yaml'
):
yield Manifest.from_file(str(mf), pkg_manager=pkg_manager)
@staticmethod
def by_config(
conf_file: Optional[str] = None,
pkg_manager: Optional[str] = None,
) -> Generator[Manifest, None, None]:
"""
Get all the manifest objects associated to the extensions declared in a
given configuration file.
"""
import platypush
from platypush.config import Config
conf_args = []
if conf_file:
conf_args.append(conf_file)
Config.init(*conf_args)
app_dir = os.path.dirname(inspect.getfile(platypush))
for name in Config.get_backends().keys():
yield Manifest.from_file(
os.path.join(app_dir, 'backend', *name.split('.'), 'manifest.yaml'),
pkg_manager=pkg_manager,
)
for name in Config.get_plugins().keys():
yield Manifest.from_file(
os.path.join(app_dir, 'plugins', *name.split('.'), 'manifest.yaml'),
pkg_manager=pkg_manager,
)
def get_available_package_manager() -> Optional[str]:
"""
Get the name of the available package manager on the system, if supported.
"""
# pylint: disable=global-statement
global _available_package_manager
if _available_package_manager:
return _available_package_manager
available_package_managers = [
pkg_manager
for pkg_manager in supported_package_managers
if shutil.which(pkg_manager)
]
if not available_package_managers:
logger.warning(
'\nYour OS does not provide any of the supported package managers.\n'
'You may have to install some optional dependencies manually.\n'
'Supported package managers: %s.\n',
', '.join(supported_package_managers.keys()),
)
return None
_available_package_manager = available_package_managers[0]
return _available_package_manager
_manifest_class_by_type: Mapping[ManifestType, Type[Manifest]] = {
@ -104,97 +333,3 @@ _manifest_class_by_type: Mapping[ManifestType, Type[Manifest]] = {
_manifest_type_by_class: Mapping[Type[Manifest], ManifestType] = {
cls: t for t, cls in _manifest_class_by_type.items()
}
def scan_manifests(base_class: Type) -> Iterable[str]:
for mf in pathlib.Path(os.path.dirname(inspect.getfile(base_class))).rglob('manifest.yaml'):
yield str(mf)
def get_manifests(base_class: Type) -> Iterable[Manifest]:
return [
Manifest.from_file(mf)
for mf in scan_manifests(base_class)
]
def get_components(base_class: Type) -> Iterable:
manifests = get_manifests(base_class)
components = {mf.get_component() for mf in manifests}
return {comp for comp in components if comp is not None}
def get_manifests_from_conf(conf_file: Optional[str] = None) -> Mapping[str, Manifest]:
import platypush
from platypush.config import Config
conf_args = []
if conf_file:
conf_args.append(conf_file)
Config.init(*conf_args)
app_dir = os.path.dirname(inspect.getfile(platypush))
manifest_files = set()
for name in Config.get_backends().keys():
manifest_files.add(os.path.join(app_dir, 'backend', *name.split('.'), 'manifest.yaml'))
for name in Config.get_plugins().keys():
manifest_files.add(os.path.join(app_dir, 'plugins', *name.split('.'), 'manifest.yaml'))
return {
manifest_file: Manifest.from_file(manifest_file)
for manifest_file in manifest_files
}
def get_dependencies_from_conf(conf_file: Optional[str] = None) -> Mapping[str, Iterable[str]]:
manifests = get_manifests_from_conf(conf_file)
deps = {
'pip': set(),
'packages': set(),
'exec': set(),
}
for manifest in manifests.values():
deps['pip'].update(manifest.install.get('pip', set()))
deps['exec'].update(manifest.install.get('exec', set()))
has_requires_packages = len([
section for section in manifest.install.keys()
if section in supported_package_managers
]) > 0
if has_requires_packages:
pkg_manager = get_available_package_manager()
deps['packages'].update(manifest.install.get(pkg_manager, set()))
return deps
def get_install_commands_from_conf(conf_file: Optional[str] = None) -> Mapping[str, str]:
deps = get_dependencies_from_conf(conf_file)
return {
'pip': f'pip install {" ".join(deps["pip"])}',
'exec': deps["exec"],
'packages': f'{supported_package_managers[_available_package_manager]} {" ".join(deps["packages"])}'
if deps['packages'] else None,
}
def get_available_package_manager() -> str:
global _available_package_manager
if _available_package_manager:
return _available_package_manager
available_package_managers = [
pkg_manager for pkg_manager in supported_package_managers.keys()
if shutil.which(pkg_manager)
]
assert available_package_managers, (
'Your OS does not provide any of the supported package managers. '
f'Supported package managers: {supported_package_managers.keys()}'
)
_available_package_manager = available_package_managers[0]
return _available_package_manager

1
setup.cfg
View File

@ -11,5 +11,6 @@ max-line-length = 120
extend-ignore =
E203
W503
SIM104
SIM105