def _get_plugins()

in submitit/core/plugins.py [0:0]

• 27 lines of code
• 9 McCabe index (conditional complexity)

def _get_plugins() -> Tuple[List[Type["Executor"]], List["JobEnvironment"]]:
    # pylint: disable=cyclic-import,import-outside-toplevel
    # Load dynamically to avoid import cycle
    # pkg_resources goes through all modules on import.
    import pkg_resources

    from ..local import debug, local
    from ..slurm import slurm

    # TODO: use sys.modules.keys() and importlib.resources to find the files
    # We load both kind of entry points at the same time because we have to go through all module files anyway.
    executors: List[Type["Executor"]] = [slurm.SlurmExecutor, local.LocalExecutor, debug.DebugExecutor]
    job_envs = [slurm.SlurmJobEnvironment(), local.LocalJobEnvironment(), debug.DebugJobEnvironment()]
    for entry_point in pkg_resources.iter_entry_points("submitit"):
        if entry_point.name not in ("executor", "job_environment"):
            logger.warning(f"Found unknown entry point in package {entry_point.module_name}: {entry_point}")
            continue

        try:
            # call `load` rather than `resolve`.
            # `load` also checks the module and its dependencies are correctly installed.
            cls = entry_point.load()
        except Exception as e:
            # This may happen if the plugin haven't been correctly installed
            logger.exception(f"Failed to load submitit plugin '{entry_point.module_name}': {e}")
            continue

        if entry_point.name == "executor":
            executors.append(cls)
        else:
            try:
                job_env = cls()
            except Exception as e:
                logger.exception(
                    f"Failed to init JobEnvironment '{cls.name}' ({cls}) from submitit plugin '{entry_point.module_name}': {e}"
                )
                continue
            job_envs.append(job_env)

    return (executors, job_envs)