Source code for laborchestrator.pythonlab_process_finder

"""
This module provides the ProcessFinder.
A tool ta get a list of all available PythonLan processes in a specified directory or module.
"""
import importlib
import inspect
import pkgutil
from contextlib import suppress
from types import ModuleType
from typing import NamedTuple, List

from pythonlab.process import PLProcess




class ImportableProcess(NamedTuple):
    module: ModuleType
    name: str





class ProcessFinder:


    @staticmethod
    def get_processes(pck: ModuleType) -> List[ImportableProcess]:
        process_names = ProcessFinder._find_processes(pck)
        return process_names




    @staticmethod
    def _find_processes(pck: ModuleType) -> List[ImportableProcess]:
        """
        Iterate through the given process package and find PLProcess classes.

        Specifying the second argument (prefix) to iter_modules makes the
        returned name an absolute name instead of a relative one. This allows
        import_module to work without having to do additional modification to
        the name.
        s. https://packaging.python.org/guides/creating-and-discovering-plugins/
        """
        processes = []
        for _finder, name, ispkg in pkgutil.iter_modules(pck.__path__):
            if ispkg:
                continue

            mod_name = pck.__name__ + "." + name
            try:
                submodule = importlib.import_module(mod_name)
            except Exception:
                continue

            for attr, obj in vars(submodule).items():
                with suppress(Exception):
                    if ProcessFinder._is_importable_process_type(obj, submodule):
                        processes.append(ImportableProcess(submodule, attr))
        return processes




    @staticmethod
    def _is_importable_process_type(obj: object, module: ModuleType) -> bool:
        if not inspect.isclass(obj):
            return False
        if obj is PLProcess:
            return False
        if inspect.isabstract(obj):
            return False
        if obj.__module__ != module.__name__:
            return False
        if not issubclass(obj, PLProcess):
            return False
        return ProcessFinder._can_be_constructed_without_args(obj)




    @staticmethod
    def _can_be_constructed_without_args(process_type: type[PLProcess]) -> bool:
        try:
            signature = inspect.signature(process_type)
        except (TypeError, ValueError):
            return False

        for parameter in signature.parameters.values():
            if parameter.kind in (
                inspect.Parameter.VAR_POSITIONAL,
                inspect.Parameter.VAR_KEYWORD,
            ):
                continue
            if parameter.default is inspect.Parameter.empty:
                return False
        return True




    @staticmethod
    def create_process(importable_process: ImportableProcess) -> PLProcess:
        process = getattr(importable_process.module, importable_process.name)()
        assert isinstance(process, PLProcess)
        return process