Source code for jenni.models.folderbase

import logging
import sys
from abc import ABC
from importlib import import_module
from io import TextIOWrapper
from pathlib import Path
from types import ModuleType
from typing import Type, Optional

from jenni.models import itembase
from jenni.models.itembase import ItemBase
from jenni.utils import tidy_text, quote3xs


class FolderBase(ItemBase, ABC):
    def yield_items(self, module_name_to_load: str = ""):
        """
        Yield this folder and any items generated by python files below this folder.
        Overwrite in subclasses to dynamically generate items, but always insert this:
        ``yield from super().yield_items(module_name_to_load)``
        """
        # logging.debug(f"yield_items({self}, '{module_name_to_load}'")
        yield from super().yield_items(module_name_to_load)
        module: ModuleType = sys.modules[self.__module__]
        # For dynamically generated modules we do not scan for nested .py files.
        if getattr(module, "is_dynamically_generated", False):
            return
        py_file: Path
        folder_path = Path(module.__file__).parent
        for py_file in list(folder_path.glob("*.py")) + list(folder_path.glob("*/__init__.py")):
            if py_file.name == "__init__.py":
                if py_file.parent == folder_path:
                    continue
                item_name = py_file.parent.name
                module_name = f"{self.__module__}.{py_file.parent.name}.{py_file.name}"[:-12]
            else:
                item_name = py_file.name[:-3]
                module_name = f"{self.__module__}.{py_file.name}"[:-3]
            logging.debug(f"Importing module {module_name}")
            module = import_module(module_name)
            item: Optional[ItemBase] = itembase.item_by_module.get(module_name)
            if item is None:
                module.__parent_item__ = self
                module.__item_name__ = item_name
                if module.__file__.endswith("__init__.py"):
                    if hasattr(module, "Folder"):
                        item = module.Folder()
                        item.parent.add_new_folder(item)
                    else:
                        raise SystemExit(f"Module {module} should contain Folder class")
                else:
                    if hasattr(module, "Job"):
                        item = module.Job()
                        item.parent.add_new_job(item)
                    else:
                        raise SystemExit(f"Module {module} should contain Job class")
            yield from item.yield_items()

    def add_new_job(self, job: "jenni.models.jobbase.JobBase"):
        """
        Called after a job has been added to this folder.
        The job can be customised by overriding this method.
        """
        pass

    def add_new_folder(self, sub_folder: "jenni.models.folderbase.FolderBase"):
        """
        Called after a sub-folder has been added to this folder.
        The sub-folder can be customised by overriding this method.
        Note yield_items() is not yet called on this job.
        """
        pass

    def yield_items_pre_sub_item(self, sub_item: ItemBase):
        """
        Called before sub_item.yield_items() is called.
        :param sub_item: the sub-item of this folder, just before calling yield_items on it.
        """
        yield from ()

    def yield_items_post_sub_item(self, sub_item: ItemBase):
        """
        Called after sub_item.yield_items() has been called.
        :param sub_item: the sub-item of this folder, just after calling yield_items on it.
        """
        yield from ()

    def sub_item_completed_handler(self, sub_item: ItemBase):
        """
        Called after a sub-item has been added to this folder
        and all yielding has completed for it.
        The item can be customised by overriding this method.
        """
        pass

    def write_jobdsl(self, fp: TextIOWrapper):
        logging.debug(f"write_jobdsl start {self}")
        # Root folder doesn't need creating:
        if self.parent is None:
            return
        jobdsl = [
            "folder('" + self.jenkins_path + "') {",
            f"    displayName({quote3xs(self.name if self.title == '' else self.title)})",
            f"""    description({quote3xs(tidy_text(self.description))})""",
            "}",
        ]

        fp.write("\n".join(jobdsl))