Source code for plynx.utils.plugin_manager

"""Utils that materialize plugins"""
# pylint: disable=global-statement
import logging
import pydoc
from typing import Any, Dict

from plynx.base.executor import BaseExecutor
from plynx.base.hub import BaseHub
from plynx.base.resource import BaseResource
from plynx.utils.config import PluginsConfig, get_plugins


[docs]def _isinstance_namedtuple(x: Any) -> bool: return isinstance(x, tuple) and getattr(x, '_fields', None) is not None
[docs]def _as_dict(obj: Any): if _isinstance_namedtuple(obj): return { key: _as_dict(getattr(obj, key)) for key in obj._fields } elif isinstance(obj, list): return [_as_dict(value) for value in obj] return obj
[docs]class _ResourceManager: def __init__(self, plugins: PluginsConfig): self.resources = plugins.resources self.kind_to_resource_class: Dict[str, BaseResource] = { resource.kind: pydoc.locate(resource.cls) # type: ignore for resource in self.resources } self.kind_to_resource_dict: Dict[str, Dict[str, Any]] = { resource.kind: resource._asdict() for resource in self.resources } for kind, resource_class in self.kind_to_resource_class.items(): logging.info(f"Initializing resource `{kind}`") self.kind_to_resource_dict[kind]['display_raw'] = resource_class.DISPLAY_RAW
[docs]class _ExecutorManager: def __init__(self, plugins: PluginsConfig): self.kind_to_executor_class: Dict[str, BaseExecutor] = {} self.kind_to_icon = {} self.kind_to_color = {} self.kind_to_title = {} for o_or_w in plugins.workflows + plugins.operations + plugins.dummy_operations: self.kind_to_executor_class[o_or_w.kind] = pydoc.locate(o_or_w.executor) # type: ignore if not self.kind_to_executor_class[o_or_w.kind]: raise Exception(f"Executor `{o_or_w.executor}` not found") self.kind_to_icon[o_or_w.kind] = o_or_w.icon self.kind_to_color[o_or_w.kind] = o_or_w.color self.kind_to_title[o_or_w.kind] = o_or_w.title self.kind_info = {} for kind, executor_class in self.kind_to_executor_class.items(): logging.info(f"Initializing executor `{kind}`") self.kind_info[kind] = { 'is_graph': executor_class.IS_GRAPH, 'title': self.kind_to_title[kind], 'icon': self.kind_to_icon[kind], 'color': self.kind_to_color[kind],
}
[docs]class _OperationManager: def __init__(self, plugins: PluginsConfig): self.operations = plugins.operations self.kind_to_operation_dict = { operation.kind: _as_dict(operation) for operation in self.operations
}
[docs]class _HubManager: def __init__(self, plugins: PluginsConfig): self.hubs = plugins.hubs self.kind_to_hub_class: Dict[str, BaseHub] = { hub.kind: pydoc.locate(hub.cls)(**hub.args) # type: ignore for hub in self.hubs } self.kind_to_hub_dict = { hub.kind: hub._asdict() for hub in self.hubs
}
[docs]class _WorkflowManager: def __init__(self, plugins: PluginsConfig): self.workflows = plugins.workflows self.kind_to_workflow_dict = { workflow.kind: workflow._asdict() for workflow in self.workflows
}
[docs]_plugins: PluginsConfig = get_plugins()
[docs]_RESOURCE_MANAGER = None
[docs]_OPERATION_MANAGER = None
[docs]_HUB_MANAGER = None
[docs]_WORKFLOW_MANAGER = None
[docs]_EXECUTOR_MANAGER = None
[docs]_PLUGINS_DICT = None
[docs]def get_resource_manager(): """Generate Resource plugin structure""" global _RESOURCE_MANAGER if not _RESOURCE_MANAGER: _RESOURCE_MANAGER = _ResourceManager(_plugins) return _RESOURCE_MANAGER
[docs]def get_operation_manager(): """Generate Operation plugin structure""" global _OPERATION_MANAGER if not _OPERATION_MANAGER: _OPERATION_MANAGER = _OperationManager(_plugins) return _OPERATION_MANAGER
[docs]def get_hub_manager(): """Generate Hub plugin structure""" global _HUB_MANAGER if not _HUB_MANAGER: _HUB_MANAGER = _HubManager(_plugins) return _HUB_MANAGER
[docs]def get_workflow_manager(): """Generate Workflow plugin structure""" global _WORKFLOW_MANAGER if not _WORKFLOW_MANAGER: _WORKFLOW_MANAGER = _WorkflowManager(_plugins) return _WORKFLOW_MANAGER
[docs]def get_executor_manager(): """Generate Exectutor plugin structure""" global _EXECUTOR_MANAGER if not _EXECUTOR_MANAGER: _EXECUTOR_MANAGER = _ExecutorManager(_plugins) return _EXECUTOR_MANAGER
[docs]def get_plugins_dict(): """Generate all of the plugins structure""" global _PLUGINS_DICT if not _PLUGINS_DICT: _PLUGINS_DICT = { 'resources_dict': get_resource_manager().kind_to_resource_dict, 'operations_dict': get_operation_manager().kind_to_operation_dict, 'hubs_dict': get_hub_manager().kind_to_hub_dict, 'workflows_dict': get_workflow_manager().kind_to_workflow_dict, 'executors_info': get_executor_manager().kind_info, } return _PLUGINS_DICT