From 7db2d139374540d775804031fbe3cd944569e96c Mon Sep 17 00:00:00 2001 From: Vsevolod Fedorov Date: Mon, 20 Mar 2023 11:08:27 +0300 Subject: [PATCH] Refactor ModuleRegistry.dispatch method Extract _load_eps and _dispatch_macro methods. Change-Id: I6c87229a16e0198f11d199c60f5cf1a5d46770fb --- jenkins_jobs/registry.py | 208 ++++++++++++++++++++------------------- 1 file changed, 107 insertions(+), 101 deletions(-) diff --git a/jenkins_jobs/registry.py b/jenkins_jobs/registry.py index c1babc426..f9ea22271 100644 --- a/jenkins_jobs/registry.py +++ b/jenkins_jobs/registry.py @@ -236,112 +236,16 @@ class ModuleRegistry(object): # Look for a component function defined in an entry point eps = self._entry_points_cache.get(component_list_type) if eps is None: - logging.debug("Caching entrypoints for %s" % component_list_type) - module_eps = [] - # auto build entry points by inferring from base component_types - mod = pkg_resources.EntryPoint( - "__all__", entry_point.module_name, dist=entry_point.dist + eps = self._load_eps( + component_list_type, component_type, entry_point, eps, name ) - Mod = mod.load() - func_eps = [ - Mod.__dict__.get(a) - for a in dir(Mod) - if isinstance(Mod.__dict__.get(a), types.FunctionType) - ] - for func_ep in func_eps: - try: - # extract entry point based on docstring - name_line = func_ep.__doc__.split("\n") - if not name_line[0].startswith("yaml:"): - logger.debug("Ignoring '%s' as an entry point" % name_line) - continue - ep_name = name_line[0].split(" ")[1] - except (AttributeError, IndexError): - # AttributeError by docstring not being defined as - # a string to have split called on it. - # IndexError raised by name_line not containing anything - # after the 'yaml:' string. - logger.debug( - "Not including func '%s' as an entry point" % func_ep.__name__ - ) - continue - - module_eps.append( - pkg_resources.EntryPoint( - ep_name, - entry_point.module_name, - dist=entry_point.dist, - attrs=(func_ep.__name__,), - ) - ) - logger.debug( - "Adding auto EP '%s=%s:%s'" - % (ep_name, entry_point.module_name, func_ep.__name__) - ) - - # load from explicitly defined entry points - module_eps.extend( - list( - pkg_resources.iter_entry_points( - group="jenkins_jobs.{0}".format(component_list_type) - ) - ) - ) - - eps = {} - for module_ep in module_eps: - if module_ep.name in eps: - raise JenkinsJobsException( - "Duplicate entry point found for component type: " - "'{0}', '{0}'," - "name: '{1}'".format(component_type, name) - ) - - eps[module_ep.name] = module_ep.load() - - # cache both sets of entry points - self._entry_points_cache[component_list_type] = eps - logger.debug("Cached entry point group %s = %s", component_list_type, eps) - macro_dict = self.macros.get(component_type, {}) macro = macro_dict.get(name) if macro: - if name in eps and name not in self.masked_warned: - self.masked_warned[name] = True - logger.warning( - "You have a macro ('%s') defined for '%s' " - "component type that is masking an inbuilt " - "definition" % (name, component_type) - ) - - # Expand macro strings only if at least one macro parameter is provided. - if component_data: - expander = self._params_expander - else: - expander = self._expander - expander_params = {**component_data, **(job_data or {})} - - elements = macro.elements - if isinstance(elements, BaseYamlObject): - # Expand !j2-yaml element right below macro body. - elements = elements.expand(expander, expander_params) - - for b in elements: - try: - element = expander.expand(b, expander_params) - except JenkinsJobsException as x: - raise JenkinsJobsException(f"While expanding macro {name!r}: {x}") - # Pass component_data in as template data to this function - # so that if the macro is invoked with arguments, - # the arguments are interpolated into the real defn. - self.dispatch( - component_type, - xml_parent, - element, - component_data, - job_data=job_data, - ) + self._dispatch_macro( + component_data, component_type, eps, job_data, macro, name, xml_parent + ) elif name in eps: func = eps[name] kwargs = self._filter_kwargs(func, job_data=job_data) @@ -351,3 +255,105 @@ class ModuleRegistry(object): "Unknown entry point or macro '{0}' " "for component type: '{1}'.".format(name, component_type) ) + + def _dispatch_macro( + self, component_data, component_type, eps, job_data, macro, name, xml_parent + ): + if name in eps and name not in self.masked_warned: + self.masked_warned[name] = True + logger.warning( + "You have a macro ('%s') defined for '%s' " + "component type that is masking an inbuilt " + "definition" % (name, component_type) + ) + # Expand macro strings only if at least one macro parameter is provided. + if component_data: + expander = self._params_expander + else: + expander = self._expander + expander_params = {**component_data, **(job_data or {})} + elements = macro.elements + if isinstance(elements, BaseYamlObject): + # Expand !j2-yaml element right below macro body. + elements = elements.expand(expander, expander_params) + for b in elements: + try: + element = expander.expand(b, expander_params) + except JenkinsJobsException as x: + raise JenkinsJobsException(f"While expanding macro {name!r}: {x}") + # Pass component_data in as template data to this function + # so that if the macro is invoked with arguments, + # the arguments are interpolated into the real defn. + self.dispatch( + component_type, + xml_parent, + element, + component_data, + job_data=job_data, + ) + + def _load_eps(self, component_list_type, component_type, entry_point, eps, name): + logging.debug("Caching entrypoints for %s" % component_list_type) + module_eps = [] + # auto build entry points by inferring from base component_types + mod = pkg_resources.EntryPoint( + "__all__", entry_point.module_name, dist=entry_point.dist + ) + Mod = mod.load() + func_eps = [ + Mod.__dict__.get(a) + for a in dir(Mod) + if isinstance(Mod.__dict__.get(a), types.FunctionType) + ] + for func_ep in func_eps: + try: + # extract entry point based on docstring + name_line = func_ep.__doc__.split("\n") + if not name_line[0].startswith("yaml:"): + logger.debug("Ignoring '%s' as an entry point" % name_line) + continue + ep_name = name_line[0].split(" ")[1] + except (AttributeError, IndexError): + # AttributeError by docstring not being defined as + # a string to have split called on it. + # IndexError raised by name_line not containing anything + # after the 'yaml:' string. + logger.debug( + "Not including func '%s' as an entry point" % func_ep.__name__ + ) + continue + + module_eps.append( + pkg_resources.EntryPoint( + ep_name, + entry_point.module_name, + dist=entry_point.dist, + attrs=(func_ep.__name__,), + ) + ) + logger.debug( + "Adding auto EP '%s=%s:%s'" + % (ep_name, entry_point.module_name, func_ep.__name__) + ) + # load from explicitly defined entry points + module_eps.extend( + list( + pkg_resources.iter_entry_points( + group="jenkins_jobs.{0}".format(component_list_type) + ) + ) + ) + eps = {} + for module_ep in module_eps: + if module_ep.name in eps: + raise JenkinsJobsException( + "Duplicate entry point found for component type: " + "'{0}', '{0}'," + "name: '{1}'".format(component_type, name) + ) + + eps[module_ep.name] = module_ep.load() + # cache both sets of entry points + self._entry_points_cache[component_list_type] = eps + logger.debug("Cached entry point group %s = %s", component_list_type, eps) + return eps