def _get_extension_config()

in metaflow/extension_support/__init__.py [0:0]

• 25 lines of code
• 18 McCabe index (conditional complexity)

def _get_extension_config(distribution_name, tl_pkg, extension_point, config_module):
    if config_module is not None and not config_module.endswith("__init__"):
        module_name = config_module
        # file_path below will be /root/metaflow_extensions/X/Y/mfextinit_Z.py and
        # module name is metaflow_extensions.X.Y.mfextinit_Z so if we want to strip to
        # /root/metaflow_extensions, we need to remove this number of elements from the
        # filepath
        strip_from_filepath = len(module_name.split(".")) - 1
    else:
        module_name = ".".join([EXT_PKG, tl_pkg, extension_point])
        # file_path here will be /root/metaflow_extensions/X/Y/__init__.py BUT
        # module name is metaflow_extensions.X.Y so we have a 1 off compared to the
        # previous case
        strip_from_filepath = len(module_name.split("."))

    _ext_debug("        Attempting to load '%s'" % module_name)

    extension_module = _attempt_load_module(module_name)

    if extension_module:
        # We update the path to this module. This is useful if we need to package this
        # package again. Note that in most cases, packaging happens in the outermost
        # local python environment (non Conda and not remote) so we already have the
        # root_paths set when we are initially looking for metaflow_extensions package.
        # This code allows for packaging while running inside a Conda environment or
        # remotely where the root_paths has been changed since the initial packaging.
        # This currently does not happen much.
        if _all_packages[distribution_name]["root_paths"] is None:
            file_path = getattr(extension_module, "__file__")
            if file_path:
                # Common case where this is an actual init file (mfextinit_X.py or __init__.py)
                root_paths = ["/".join(file_path.split("/")[:-strip_from_filepath])]
            else:
                # Only used for plugins.cards where the package can be a NS package. In
                # this case, __path__ will have things like /root/metaflow_extensions/X/Y
                # and module name will be metaflow_extensions.X.Y
                root_paths = [
                    "/".join(p.split("/")[: -len(module_name.split(".")) + 1])
                    for p in extension_module.__path__
                ]

            _ext_debug("Package '%s' is rooted at %s" % (distribution_name, root_paths))
            _all_packages[distribution_name]["root_paths"] = root_paths

        return MFExtModule(
            package_name=distribution_name, tl_package=tl_pkg, module=extension_module
        )
    return None