blacklight/platypush
1
0
Fork 0
forked from platypush/platypush
Code Issues Pull requests Projects Releases Packages Wiki Activity

Large refactor for the inspect plugin.

Browse source 
More common logic has been extracted and all the methods and classes
have been documented and black'd.
This commit is contained in:
Fabio Manganiello 2023-05-17 00:05:22 +02:00
parent 2cba504e3b
commit 61ea3d79e4
Signed by: blacklight
GPG key ID: D90FBA7F76362774
5 changed files with 444 additions and 139 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

385
platypush/plugins/inspect/__init__.py
View file

@ -1,24 +1,36 @@
from collections import defaultdict
import importlib
import inspect
import json
import threading
from typing import Optional
import os
import pathlib
import pickle
import pkgutil
from types import ModuleType
from typing import Callable, Dict, Generator, Optional, Type, Union
from platypush.backend import Backend
from platypush.config import Config
from platypush.plugins import Plugin, action
from platypush.message.event import Event
from platypush.message.response import Response
from platypush.utils import get_plugin_class_by_name
from platypush.utils import (
get_backend_class_by_name,
get_backend_name_by_class,
get_plugin_class_by_name,
get_plugin_name_by_class,
)
from platypush.utils.manifest import Manifest, scan_manifests
from ._context import ComponentContext
from ._model import (
BackendModel,
EventModel,
Model,
PluginModel,
ProcedureEncoder,
ResponseModel,
)
from ._serialize import ProcedureEncoder
class InspectPlugin(Plugin):
@ -28,17 +40,79 @@ class InspectPlugin(Plugin):
def __init__(self, **kwargs):
super().__init__(**kwargs)
self._plugins = {}
self._backends = {}
self._events = {}
self._responses = {}
self._plugins_lock = threading.RLock()
self._backends_lock = threading.RLock()
self._events_lock = threading.RLock()
self._responses_lock = threading.RLock()
self._components_cache_file = os.path.join(
Config.get('workdir'), # type: ignore
'components.cache', # type: ignore
)
self._components_context: Dict[type, ComponentContext] = defaultdict(
ComponentContext
)
self._components_cache: Dict[type, dict] = defaultdict(dict)
self._load_components_cache()
def _get_modules(self, parent_class: type):
for mf_file in scan_manifests(parent_class):
def _load_components_cache(self):
"""
Loads the components cache from disk.
"""
try:
with open(self._components_cache_file, 'rb') as f:
self._components_cache = pickle.load(f)
except OSError:
return
def _flush_components_cache(self):
"""
Flush the current components cache to disk.
"""
with open(self._components_cache_file, 'wb') as f:
pickle.dump(self._components_cache, f)
def _get_cached_component(
self, base_type: type, comp_type: type
) -> Optional[Model]:
"""
Retrieve a cached component's ``Model``.
:param base_type: The base type of the component (e.g. ``Plugin`` or
``Backend``).
:param comp_type: The specific type of the component (e.g.
``MusicMpdPlugin`` or ``HttpBackend``).
:return: The cached component's ``Model`` if it exists, otherwise null.
"""
return self._components_cache.get(base_type, {}).get(comp_type)
def _cache_component(
self,
base_type: type,
comp_type: type,
model: Model,
index_by_module: bool = False,
):
"""
Cache the ``Model`` object for a component.
:param base_type: The base type of the component (e.g. ``Plugin`` or
``Backend``).
:param comp_type: The specific type of the component (e.g.
``MusicMpdPlugin`` or ``HttpBackend``).
:param model: The ``Model`` object to cache.
:param index_by_module: If ``True``, the ``Model`` object will be
indexed according to the ``base_type -> module -> comp_type``
mapping, otherwise ``base_type -> comp_type``.
"""
if index_by_module:
if not self._components_cache.get(base_type, {}).get(model.package):
self._components_cache[base_type][model.package] = {}
self._components_cache[base_type][model.package][comp_type] = model
else:
self._components_cache[base_type][comp_type] = model
def _scan_integrations(self, base_type: type):
"""
A generator that scans the manifest files given a ``base_type``
(``Plugin`` or ``Backend``) and yields the parsed submodules.
"""
for mf_file in scan_manifests(base_type):
manifest = Manifest.from_file(mf_file)
try:
yield importlib.import_module(manifest.package)
@ -50,118 +124,242 @@ class InspectPlugin(Plugin):
)
continue
def _scan_modules(self, base_type: type) -> Generator[ModuleType, None, None]:
"""
A generator that scan the modules given a ``base_type`` (e.g. ``Event``).
Unlike :meth:`._scan_integrations`, this method recursively scans the
modules using ``pkgutil`` instead of using the information provided in
the integrations' manifest files.
"""
prefix = base_type.__module__ + '.'
path = str(pathlib.Path(inspect.getfile(base_type)).parent)
for _, modname, _ in pkgutil.walk_packages(
path=[path], prefix=prefix, onerror=lambda _: None
):
try:
yield importlib.import_module(modname)
except Exception as e:
self.logger.debug('Could not import module %s: %s', modname, e)
continue
def _init_component(
self,
base_type: type,
comp_type: type,
model_type: Type[Model],
index_by_module: bool = False,
) -> Model:
"""
Initialize a component's ``Model`` object and cache it.
:param base_type: The base type of the component (e.g. ``Plugin`` or
``Backend``).
:param comp_type: The specific type of the component (e.g.
``MusicMpdPlugin`` or ``HttpBackend``).
:param model_type: The type of the ``Model`` object that should be
created.
:param index_by_module: If ``True``, the ``Model`` object will be
indexed according to the ``base_type -> module -> comp_type``
mapping, otherwise ``base_type -> comp_type``.
:return: The initialized component's ``Model`` object.
"""
prefix = base_type.__module__ + '.'
comp_file = inspect.getsourcefile(comp_type)
model = None
mtime = None
if comp_file:
mtime = os.stat(comp_file).st_mtime
cached_model = self._get_cached_component(base_type, comp_type)
# Only update the component model if its source file was
# modified since the last time it was scanned
if (
cached_model
and cached_model.last_modified
and mtime <= cached_model.last_modified
):
model = cached_model
if not model:
self.logger.info('Scanning component %s', comp_type.__name__)
model = model_type(comp_type, prefix=prefix, last_modified=mtime)
self._cache_component(
base_type, comp_type, model, index_by_module=index_by_module
)
return model
def _init_modules(
self,
base_type: type,
model_type: Type[Model],
):
"""
Initializes, parses and caches all the components of a given type.
Unlike :meth:`._scan_integrations`, this method inspects all the
members of a ``module`` for those that match the given ``base_type``
instead of relying on the information provided in the manifest.
It is a bit more inefficient, but it works fine for simple components
(like entities and messages) that don't require extra recursive parsing
logic for their docs (unlike plugins).
"""
for module in self._scan_modules(base_type):
for _, obj_type in inspect.getmembers(module):
if (
inspect.isclass(obj_type)
and issubclass(obj_type, base_type)
# Exclude the base_type itself
and obj_type != base_type
):
self._init_component(
base_type=base_type,
comp_type=obj_type,
model_type=model_type,
index_by_module=True,
)
def _init_integrations(
self,
base_type: Type[Union[Plugin, Backend]],
model_type: Type[Union[PluginModel, BackendModel]],
class_by_name: Callable[[str], Optional[type]],
):
"""
Initializes, parses and caches all the integrations of a given type.
:param base_type: The base type of the component (e.g. ``Plugin`` or
``Backend``).
:param model_type: The type of the ``Model`` objects that should be
created.
:param class_by_name: A function that returns the class of a given
integration given its qualified name.
"""
for module in self._scan_integrations(base_type):
comp_name = '.'.join(module.__name__.split('.')[2:])
comp_type = class_by_name(comp_name)
if not comp_type:
continue
self._init_component(
base_type=base_type,
comp_type=comp_type,
model_type=model_type,
)
self._flush_components_cache()
def _init_plugins(self):
prefix = Plugin.__module__ + '.'
for module in self._get_modules(Plugin):
plugin_name = '.'.join(module.__name__.split('.')[2:])
plugin_class = get_plugin_class_by_name(plugin_name)
model = PluginModel(plugin=plugin_class, prefix=prefix)
if model.name:
self._plugins[model.name] = model
"""
Initializes and caches all the available plugins.
"""
self._init_integrations(
base_type=Plugin,
model_type=PluginModel,
class_by_name=get_plugin_class_by_name,
)
def _init_backends(self):
prefix = Backend.__module__ + '.'
for module in self._get_modules(Backend):
for _, obj in inspect.getmembers(module):
if inspect.isclass(obj) and issubclass(obj, Backend):
model = BackendModel(backend=obj, prefix=prefix)
if model.name:
self._backends[model.name] = model
"""
Initializes and caches all the available backends.
"""
self._init_integrations(
base_type=Backend,
model_type=BackendModel,
class_by_name=get_backend_class_by_name,
)
def _init_events(self):
prefix = Event.__module__ + '.'
for module in self._get_modules(Event):
for _, obj in inspect.getmembers(module):
if type(obj) == Event: # pylint: disable=unidiomatic-typecheck
continue
if inspect.isclass(obj) and issubclass(obj, Event) and obj != Event:
event = EventModel(event=obj, prefix=prefix)
if event.package not in self._events:
self._events[event.package] = {event.name: event}
else:
self._events[event.package][event.name] = event
"""
Initializes and caches all the available events.
"""
self._init_modules(
base_type=Event,
model_type=EventModel,
)
def _init_responses(self):
prefix = Response.__module__ + '.'
"""
Initializes and caches all the available responses.
"""
self._init_modules(
base_type=Response,
model_type=ResponseModel,
)
for module in self._get_modules(Response):
for _, obj in inspect.getmembers(module):
if type(obj) == Response: # pylint: disable=unidiomatic-typecheck
continue
if (
inspect.isclass(obj)
and issubclass(obj, Response)
and obj != Response
):
response = ResponseModel(response=obj, prefix=prefix)
if response.package not in self._responses:
self._responses[response.package] = {response.name: response}
else:
self._responses[response.package][response.name] = response
def _init_components(self, base_type: type, initializer: Callable[[], None]):
"""
Context manager boilerplate for the other ``_init_*`` methods.
"""
ctx = self._components_context[base_type]
with ctx.init_lock:
if not ctx.refreshed.is_set():
initializer()
ctx.refreshed.set()
@action
def get_all_plugins(self):
"""
Get information about all the available plugins.
"""
with self._plugins_lock:
if not self._plugins:
self._init_plugins()
return json.dumps(
{name: dict(plugin) for name, plugin in self._plugins.items()}
)
self._init_components(Plugin, self._init_plugins)
return json.dumps(
{
get_plugin_name_by_class(cls): dict(plugin)
for cls, plugin in self._components_cache.get(Plugin, {}).items()
}
)
@action
def get_all_backends(self):
"""
Get information about all the available backends.
"""
with self._backends_lock:
if not self._backends:
self._init_backends()
return json.dumps(
{name: dict(backend) for name, backend in self._backends.items()}
)
self._init_components(Backend, self._init_backends)
return json.dumps(
{
get_backend_name_by_class(cls): dict(backend)
for cls, backend in self._components_cache.get(Backend, {}).items()
}
)
@action
def get_all_events(self):
"""
Get information about all the available events.
"""
with self._events_lock:
if not self._events:
self._init_events()
return json.dumps(
{
package: {name: dict(event) for name, event in events.items()}
for package, events in self._events.items()
self._init_components(Event, self._init_events)
return json.dumps(
{
package: {
obj_type.__name__: dict(event_model)
for obj_type, event_model in events.items()
}
)
for package, events in self._components_cache.get(Event, {}).items()
}
)
@action
def get_all_responses(self):
"""
Get information about all the available responses.
"""
with self._responses_lock:
if not self._responses:
self._init_responses()
return json.dumps(
{
package: {name: dict(event) for name, event in responses.items()}
for package, responses in self._responses.items()
self._init_components(Response, self._init_responses)
return json.dumps(
{
package: {
obj_type.__name__: dict(response_model)
for obj_type, response_model in responses.items()
}
)
for package, responses in self._components_cache.get(
Response, {}
).items()
}
)
@action
def get_procedures(self) -> dict:
@ -181,8 +379,7 @@ class InspectPlugin(Plugin):
if entry:
return Config.get(entry)
cfg = Config.get()
return cfg
return Config.get()
# vim:sw=4:ts=4:et:

12
platypush/plugins/inspect/_context.py Normal file
View file

@ -0,0 +1,12 @@
from dataclasses import dataclass, field
import threading
@dataclass
class ComponentContext:
"""
This class is used to store the context of a component type.
"""
init_lock: threading.RLock = field(default_factory=threading.RLock)
refreshed: threading.Event = field(default_factory=threading.Event)

131
platypush/plugins/inspect/_model.py
View file

@ -1,90 +1,131 @@
from abc import ABC, abstractmethod
import inspect
import json
import re
from typing import Optional, Type
from platypush.backend import Backend
from platypush.message.event import Event
from platypush.message.response import Response
from platypush.plugins import Plugin
from platypush.utils import get_decorators
class Model(ABC):
class Model:
"""
Base class for component models.
"""
def __init__(
self,
obj_type: type,
name: Optional[str] = None,
doc: Optional[str] = None,
prefix: str = '',
last_modified: Optional[float] = None,
) -> None:
"""
:param obj_type: Type of the component.
:param name: Name of the component.
:param doc: Documentation of the component.
:param last_modified: Last modified timestamp of the component.
"""
self._obj_type = obj_type
self.package = obj_type.__module__[len(prefix) :]
self.name = name or self.package
self.doc = doc or obj_type.__doc__
self.last_modified = last_modified
def __str__(self):
"""
:return: JSON string representation of the model.
"""
return json.dumps(dict(self), indent=2, sort_keys=True)
def __repr__(self):
"""
:return: JSON string representation of the model.
"""
return json.dumps(dict(self))
@abstractmethod
def __iter__(self):
raise NotImplementedError()
class ProcedureEncoder(json.JSONEncoder):
def default(self, o):
if callable(o):
return {
'type': 'native_function',
}
return super().default(o)
class BackendModel(Model):
def __init__(self, backend, prefix=''):
self.name = backend.__module__[len(prefix) :]
self.doc = backend.__doc__
def __iter__(self):
"""
Iterator for the model public attributes/values pairs.
"""
for attr in ['name', 'doc']:
yield attr, getattr(self, attr)
# pylint: disable=too-few-public-methods
class BackendModel(Model):
"""
Model for backend components.
"""
def __init__(self, obj_type: Type[Backend], *args, **kwargs):
super().__init__(obj_type, *args, **kwargs)
# pylint: disable=too-few-public-methods
class PluginModel(Model):
def __init__(self, plugin, prefix=''):
self.name = re.sub(r'\._plugin$', '', plugin.__module__[len(prefix) :])
self.doc = plugin.__doc__
"""
Model for plugin components.
"""
def __init__(self, obj_type: Type[Plugin], prefix: str = '', **kwargs):
super().__init__(
obj_type,
name=re.sub(r'\._plugin$', '', obj_type.__module__[len(prefix) :]),
**kwargs,
)
self.actions = {
action_name: ActionModel(getattr(plugin, action_name))
for action_name in get_decorators(plugin, climb_class_hierarchy=True).get(
action_name: ActionModel(getattr(obj_type, action_name))
for action_name in get_decorators(obj_type, climb_class_hierarchy=True).get(
'action', []
)
}
def __iter__(self):
"""
Overrides the default implementation of ``__iter__`` to also include
plugin actions.
"""
for attr in ['name', 'actions', 'doc']:
if attr == 'actions':
yield attr, {
name: dict(action) for name, action in self.actions.items()
},
}
else:
yield attr, getattr(self, attr)
class EventModel(Model):
def __init__(self, event, prefix=''):
self.package = event.__module__[len(prefix) :]
self.name = event.__name__
self.doc = event.__doc__
"""
Model for event components.
"""
def __iter__(self):
for attr in ['name', 'doc']:
yield attr, getattr(self, attr)
def __init__(self, obj_type: Type[Event], **kwargs):
super().__init__(obj_type, **kwargs)
class ResponseModel(Model):
def __init__(self, response, prefix=''):
self.package = response.__module__[len(prefix) :]
self.name = response.__name__
self.doc = response.__doc__
"""
Model for response components.
"""
def __iter__(self):
for attr in ['name', 'doc']:
yield attr, getattr(self, attr)
def __init__(self, obj_type: Type[Response], **kwargs):
super().__init__(obj_type, **kwargs)
class ActionModel(Model):
def __init__(self, action):
self.name = action.__name__
self.doc, argsdoc = self._parse_docstring(action.__doc__)
"""
Model for plugin action components.
"""
def __init__(self, action, **kwargs):
doc, argsdoc = self._parse_docstring(action.__doc__)
super().__init__(action, name=action.__name__, doc=doc, **kwargs)
self.args = {}
self.has_kwargs = False

16
platypush/plugins/inspect/_serialize.py Normal file
View file

@ -0,0 +1,16 @@
import json
class ProcedureEncoder(json.JSONEncoder):
"""
Encoder for the Procedure model.
"""
def default(self, o):
if callable(o):
return {
'type': 'native_function',
'source': f'{o.__module__}.{o.__name__}',
}
return super().default(o)

39
platypush/utils/__init__.py
View file

@ -73,6 +73,17 @@ def get_plugin_module_by_name(plugin_name):
return None
def get_backend_module_by_name(backend_name):
"""Gets the module of a backend by name (e.g. "backend.http" or "backend.mqtt")"""
module_name = 'platypush.backend.' + backend_name
try:
return importlib.import_module('platypush.backend.' + backend_name)
except ImportError as e:
logger.error('Cannot import %s: %s', module_name, e)
return None
def get_plugin_class_by_name(plugin_name):
"""Gets the class of a plugin by name (e.g. "music.mpd" or "media.vlc")"""
@ -110,6 +121,34 @@ def get_plugin_name_by_class(plugin) -> Optional[str]:
return '.'.join(class_tokens)
def get_backend_class_by_name(backend_name: str):
"""Gets the class of a backend by name (e.g. "backend.http" or "backend.mqtt")"""
module = get_backend_module_by_name(backend_name)
if not module:
return
class_name = getattr(
module,
''.join(
[
token.capitalize()
for i, token in enumerate(backend_name.split('.'))
if not (i == 0 and token == 'backend')
]
)
+ 'Backend',
)
try:
return getattr(
module,
''.join([_.capitalize() for _ in backend_name.split('.')]) + 'Backend',
)
except Exception as e:
logger.error('Cannot import class %s: %s', class_name, e)
return None
def get_backend_name_by_class(backend) -> Optional[str]:
"""Gets the common name of a backend (e.g. "http" or "mqtt") given its class."""