Function import_script_module

dlt/reflection/script_inspector.py:33–61

Loads a module in `script_relative_path` by splitting it into a script module (file part) and package (folders). `module_path` is added to sys.path Optionally, missing imports will be ignored by importing a dummy module instead.

(
    module_path: str, script_relative_path: str, ignore_missing_imports: bool = False
)

def import_script_module(
    module_path: str, script_relative_path: str, ignore_missing_imports: bool = False
) -> ModuleType:
    """Loads a module in `script_relative_path` by splitting it into a script module (file part) and package (folders).  `module_path` is added to sys.path
    Optionally, missing imports will be ignored by importing a dummy module instead.
    """
    if os.path.isabs(script_relative_path):
        raise ValueError(script_relative_path, f"Not relative path to `{module_path}`")

    module, _ = os.path.splitext(script_relative_path)
    module = ".".join(Path(module).parts)

    # add path to module search
    sys_path: str = None
    if module_path not in sys.path:
        sys_path = module_path
        # path must be first so we always load our module of
        sys.path.insert(0, sys_path)
    try:
        logger.info(f"Importing pipeline script from module `{module}` with path `{module_path}`")
        if ignore_missing_imports:
            return import_module_with_missing(module)
        else:
            return import_module(module)

    finally:
        # remove script module path
        if sys_path:
            sys.path.remove(sys_path)


def import_pipeline_script(