From 1e7cfe22309cc3cc72ed0e0c4e2203c7fec7919f Mon Sep 17 00:00:00 2001 From: Chris Sewell Date: Thu, 28 Sep 2023 11:32:16 -0300 Subject: [PATCH] =?UTF-8?q?=E2=99=BB=EF=B8=8F=20Refactor=20needs=20process?= =?UTF-8?q?ing=20function=20signatures?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- sphinx_needs/api/need.py | 3 +- sphinx_needs/builder.py | 6 ++- sphinx_needs/config.py | 2 + sphinx_needs/directives/need.py | 35 +++++--------- sphinx_needs/directives/needbar.py | 2 +- sphinx_needs/directives/needextend.py | 16 +++---- sphinx_needs/directives/needflow.py | 4 +- sphinx_needs/directives/needgantt.py | 4 +- sphinx_needs/directives/needimport.py | 5 +- sphinx_needs/directives/needpie.py | 5 +- sphinx_needs/directives/needsequence.py | 4 +- sphinx_needs/directives/needuml.py | 3 +- sphinx_needs/filter_common.py | 25 +++++----- sphinx_needs/functions/common.py | 12 +++-- sphinx_needs/functions/functions.py | 64 ++++++++++++++----------- sphinx_needs/need_constraints.py | 16 +++---- sphinx_needs/roles/need_count.py | 12 ++--- sphinx_needs/warnings.py | 10 ++-- 18 files changed, 117 insertions(+), 111 deletions(-) diff --git a/sphinx_needs/api/need.py b/sphinx_needs/api/need.py index f3415cc5b..5ae06f619 100644 --- a/sphinx_needs/api/need.py +++ b/sphinx_needs/api/need.py @@ -746,6 +746,7 @@ def _merge_global_options(app: Sphinx, needs_info, global_options) -> None: """Add all global defined options to needs_info""" if global_options is None: return + config = NeedsSphinxConfig(app.config) for key, value in global_options.items(): # If key already exists in needs_info, this global_option got overwritten manually in current need if key in needs_info and needs_info[key]: @@ -762,7 +763,7 @@ def _merge_global_options(app: Sphinx, needs_info, global_options) -> None: for single_value in values: if len(single_value) < 2 or len(single_value) > 3: raise NeedsInvalidException(f"global option tuple has wrong amount of parameters: {key}") - if filter_single_need(app, needs_info, single_value[1]): + if filter_single_need(needs_info, config, single_value[1]): # Set value, if filter has matched needs_info[key] = single_value[0] elif len(single_value) == 3 and (key not in needs_info.keys() or len(str(needs_info[key])) > 0): diff --git a/sphinx_needs/builder.py b/sphinx_needs/builder.py index 56bed77cc..454cc95b9 100644 --- a/sphinx_needs/builder.py +++ b/sphinx_needs/builder.py @@ -48,7 +48,9 @@ def finish(self) -> None: from sphinx_needs.filter_common import filter_needs filter_string = needs_config.builder_filter - filtered_needs: List[NeedsInfoType] = filter_needs(self.app, data.get_or_create_needs().values(), filter_string) + filtered_needs: List[NeedsInfoType] = filter_needs( + data.get_or_create_needs().values(), needs_config, filter_string + ) for need in filtered_needs: needs_list.add_need(version, need) @@ -181,7 +183,7 @@ def finish(self) -> None: filter_string = needs_config.builder_filter from sphinx_needs.filter_common import filter_needs - filtered_needs = filter_needs(self.app, needs, filter_string) + filtered_needs = filter_needs(needs, needs_config, filter_string) needs_build_json_per_id_path = needs_config.build_json_per_id_path needs_dir = os.path.join(self.outdir, needs_build_json_per_id_path) if not os.path.exists(needs_dir): diff --git a/sphinx_needs/config.py b/sphinx_needs/config.py index da452c424..0ba6c559f 100644 --- a/sphinx_needs/config.py +++ b/sphinx_needs/config.py @@ -87,6 +87,8 @@ def __init__(self, config: _SphinxConfig) -> None: super().__setattr__("_config", config) def __getattribute__(self, name: str) -> Any: + if name.startswith("__"): + return super().__getattribute__(name) return getattr(super().__getattribute__("_config"), f"needs_{name}") def __setattr__(self, name: str, value: Any) -> None: diff --git a/sphinx_needs/directives/need.py b/sphinx_needs/directives/need.py index 3895bdf5a..2636f403a 100644 --- a/sphinx_needs/directives/need.py +++ b/sphinx_needs/directives/need.py @@ -14,7 +14,7 @@ from sphinx_needs.api import add_need from sphinx_needs.api.exceptions import NeedsInvalidException from sphinx_needs.config import NEEDS_CONFIG, NeedsSphinxConfig -from sphinx_needs.data import SphinxNeedsData +from sphinx_needs.data import NeedsInfoType, SphinxNeedsData from sphinx_needs.debug import measure_time from sphinx_needs.defaults import NEED_DEFAULT_OPTIONS from sphinx_needs.directives.needextend import ( @@ -366,11 +366,6 @@ def process_need_nodes(app: Sphinx, doctree: nodes.document, fromdocname: str) - """ Event handler to add title meta data (status, tags, links, ...) information to the Need node. Also processes constraints. - - :param app: - :param doctree: - :param fromdocname: - :return: """ needs_config = NeedsSphinxConfig(app.config) if not needs_config.include_needs: @@ -381,18 +376,19 @@ def process_need_nodes(app: Sphinx, doctree: nodes.document, fromdocname: str) - env = app.env needs_data = SphinxNeedsData(env) + needs = needs_data.get_or_create_needs() # If no needs were defined, we do not need to do anything - if not needs_data.get_or_create_needs(): + if not needs: return if not needs_data.needs_is_post_processed: - resolve_dynamic_values(env) - resolve_variants_options(env) - check_links(env) - create_back_links(env) - process_constraints(app) - extend_needs_data(app) + resolve_dynamic_values(needs, app) + resolve_variants_options(needs, needs_config, app.builder.tags.tags) + check_links(needs, needs_config) + create_back_links(needs, needs_config) + process_constraints(needs, needs_config) + extend_needs_data(needs, needs_data.get_or_create_extends(), needs_config) needs_data.needs_is_post_processed = True for extend_node in doctree.findall(Needextend): @@ -429,16 +425,13 @@ def print_need_nodes(app: Sphinx, doctree: nodes.document, fromdocname: str, fou build_need(layout, node_need, app, fromdocname=fromdocname) -def check_links(env: BuildEnvironment) -> None: +def check_links(needs: Dict[str, NeedsInfoType], config: NeedsSphinxConfig) -> None: """Checks if set links are valid or are dead (referenced need does not exist.) For needs with dead links, an extra ``has_dead_links`` field is added and, if the link is not allowed to be dead, the ``has_forbidden_dead_links`` field is also added. """ - config = NeedsSphinxConfig(env.config) - data = SphinxNeedsData(env) - needs = data.get_or_create_needs() extra_links = config.extra_links for need in needs.values(): for link_type in extra_links: @@ -461,17 +454,13 @@ def check_links(env: BuildEnvironment) -> None: break # One found dead link is enough -def create_back_links(env: BuildEnvironment) -> None: +def create_back_links(needs: Dict[str, NeedsInfoType], config: NeedsSphinxConfig) -> None: """Create back-links in all found needs. These are fields for each link type, ``_back``, which contain a list of all IDs of needs that link to the current need. """ - data = SphinxNeedsData(env) - needs_config = NeedsSphinxConfig(env.config) - needs = data.get_or_create_needs() - - for links in needs_config.extra_links: + for links in config.extra_links: option = links["option"] option_back = f"{option}_back" diff --git a/sphinx_needs/directives/needbar.py b/sphinx_needs/directives/needbar.py index a61daab6a..1ddd5d572 100644 --- a/sphinx_needs/directives/needbar.py +++ b/sphinx_needs/directives/needbar.py @@ -271,7 +271,7 @@ def process_needbar(app: Sphinx, doctree: nodes.document, fromdocname: str, foun if element.isdigit(): line_number.append(float(element)) else: - result = len(filter_needs(app, need_list, element)) + result = len(filter_needs(need_list, needs_config, element)) line_number.append(float(result)) local_data_number.append(line_number) diff --git a/sphinx_needs/directives/needextend.py b/sphinx_needs/directives/needextend.py index 278593864..fdfc3789a 100644 --- a/sphinx_needs/directives/needextend.py +++ b/sphinx_needs/directives/needextend.py @@ -7,12 +7,11 @@ from docutils import nodes from docutils.parsers.rst import directives -from sphinx.application import Sphinx from sphinx.util.docutils import SphinxDirective from sphinx_needs.api.exceptions import NeedsInvalidFilter from sphinx_needs.config import NeedsSphinxConfig -from sphinx_needs.data import SphinxNeedsData +from sphinx_needs.data import NeedsExtendType, NeedsInfoType, SphinxNeedsData from sphinx_needs.filter_common import filter_needs from sphinx_needs.logging import get_logger from sphinx_needs.utils import add_doc @@ -71,11 +70,10 @@ def run(self) -> Sequence[nodes.Node]: return [targetnode, Needextend("")] -def extend_needs_data(app: Sphinx) -> None: +def extend_needs_data( + all_needs: Dict[str, NeedsInfoType], extends: Dict[str, NeedsExtendType], needs_config: NeedsSphinxConfig +) -> None: """Use data gathered from needextend directives to modify fields of existing needs.""" - env = app.env - needs_config = NeedsSphinxConfig(env.config) - data = SphinxNeedsData(env) list_values = ( ["tags", "links"] @@ -84,9 +82,7 @@ def extend_needs_data(app: Sphinx) -> None: ) # back-links (incoming) link_names = [x["option"] for x in needs_config.extra_links] - all_needs = data.get_or_create_needs() - - for current_needextend in data.get_or_create_extends().values(): + for current_needextend in extends.values(): need_filter = current_needextend["filter"] if need_filter in all_needs: # a single known ID @@ -102,7 +98,7 @@ def extend_needs_data(app: Sphinx) -> None: else: # a filter string try: - found_needs = filter_needs(app, all_needs.values(), need_filter) + found_needs = filter_needs(all_needs.values(), needs_config, need_filter) except NeedsInvalidFilter as e: raise NeedsInvalidFilter( f"Filter not valid for needextend on page {current_needextend['docname']}:\n{e}" diff --git a/sphinx_needs/directives/needflow.py b/sphinx_needs/directives/needflow.py index 98ead1bd6..e586b4d70 100644 --- a/sphinx_needs/directives/needflow.py +++ b/sphinx_needs/directives/needflow.py @@ -130,7 +130,9 @@ def get_need_node_rep_for_plantuml( # We set # later, as the user may not have given a color and the node must get highlighted node_colors.append(need_info["type_color"].replace("#", "")) - if current_needflow["highlight"] and filter_single_need(app, need_info, current_needflow["highlight"], all_needs): + if current_needflow["highlight"] and filter_single_need( + need_info, needs_config, current_needflow["highlight"], all_needs + ): node_colors.append("line:FF0000") # need parts style use default "rectangle" diff --git a/sphinx_needs/directives/needgantt.py b/sphinx_needs/directives/needgantt.py index abec283bd..ec6dc182d 100644 --- a/sphinx_needs/directives/needgantt.py +++ b/sphinx_needs/directives/needgantt.py @@ -214,7 +214,7 @@ def process_needgantt(app: Sphinx, doctree: nodes.document, fromdocname: str, fo complete = None if current_needgantt["milestone_filter"]: - is_milestone = filter_single_need(app, need, current_needgantt["milestone_filter"]) + is_milestone = filter_single_need(need, needs_config, current_needgantt["milestone_filter"]) else: is_milestone = False @@ -259,7 +259,7 @@ def process_needgantt(app: Sphinx, doctree: nodes.document, fromdocname: str, fo puml_node["uml"] += "\n' Constraints definition \n\n" for need in found_needs: if current_needgantt["milestone_filter"]: - is_milestone = filter_single_need(app, need, current_needgantt["milestone_filter"]) + is_milestone = filter_single_need(need, needs_config, current_needgantt["milestone_filter"]) else: is_milestone = False for con_type in ("starts_with_links", "starts_after_links", "ends_with_links"): diff --git a/sphinx_needs/directives/needimport.py b/sphinx_needs/directives/needimport.py index cdccd1409..299d718d4 100644 --- a/sphinx_needs/directives/needimport.py +++ b/sphinx_needs/directives/needimport.py @@ -123,6 +123,7 @@ def run(self) -> Sequence[nodes.Node]: if version not in needs_import_list["versions"].keys(): raise VersionNotFound(f"Version {version} not found in needs import file {correct_need_import_path}") + needs_config = NeedsSphinxConfig(self.config) # TODO this is not exactly NeedsInfoType, because the export removes/adds some keys needs_list: Dict[str, NeedsInfoType] = needs_import_list["versions"][version]["needs"] @@ -138,7 +139,7 @@ def run(self) -> Sequence[nodes.Node]: # "content" is the sphinx internal name for this kind of information filter_context["content"] = need["description"] # type: ignore[typeddict-item] try: - if filter_single_need(self.env.app, filter_context, filter_string): + if filter_single_need(filter_context, needs_config, filter_string): needs_list_filtered[key] = need except Exception as e: logger.warning( @@ -152,7 +153,7 @@ def run(self) -> Sequence[nodes.Node]: needs_list = needs_list_filtered # If we need to set an id prefix, we also need to manipulate all used ids in the imported data. - extra_links = NeedsSphinxConfig(self.config).extra_links + extra_links = needs_config.extra_links if id_prefix: for need in needs_list.values(): for id in needs_list: diff --git a/sphinx_needs/directives/needpie.py b/sphinx_needs/directives/needpie.py index 5b2659f8e..09bc3c70c 100644 --- a/sphinx_needs/directives/needpie.py +++ b/sphinx_needs/directives/needpie.py @@ -111,9 +111,10 @@ def run(self) -> Sequence[nodes.Node]: def process_needpie(app: Sphinx, doctree: nodes.document, fromdocname: str, found_nodes: List[nodes.Element]) -> None: env = app.env needs_data = SphinxNeedsData(env) + needs_config = NeedsSphinxConfig(env.config) # NEEDFLOW - include_needs = NeedsSphinxConfig(env.config).include_needs + include_needs = needs_config.include_needs # for node in doctree.findall(Needpie): for node in found_nodes: if not include_needs: @@ -146,7 +147,7 @@ def process_needpie(app: Sphinx, doctree: nodes.document, fromdocname: str, foun if line.isdigit(): sizes.append(abs(float(line))) else: - result = len(filter_needs(app, need_list, line)) + result = len(filter_needs(need_list, needs_config, line)) sizes.append(result) elif current_needpie["filter_func"] and not content: try: diff --git a/sphinx_needs/directives/needsequence.py b/sphinx_needs/directives/needsequence.py index c250921df..63ccd3aad 100644 --- a/sphinx_needs/directives/needsequence.py +++ b/sphinx_needs/directives/needsequence.py @@ -259,7 +259,9 @@ def get_message_needs( if filter: from sphinx_needs.filter_common import filter_single_need - if not filter_single_need(app, all_needs_dict[rec_id], filter, needs=all_needs_dict.values()): + if not filter_single_need( + all_needs_dict[rec_id], NeedsSphinxConfig(app.config), filter, needs=all_needs_dict.values() + ): continue rec_data = {"id": rec_id, "title": all_needs_dict[rec_id]["title"], "messages": []} diff --git a/sphinx_needs/directives/needuml.py b/sphinx_needs/directives/needuml.py index e3ea4365b..bd679cdbb 100644 --- a/sphinx_needs/directives/needuml.py +++ b/sphinx_needs/directives/needuml.py @@ -350,8 +350,9 @@ def filter(self, filter_string): """ Return a list of found needs that pass the given filter string. """ + needs_config = NeedsSphinxConfig(self.app.config) - return filter_needs(self.app, list(self.needs.values()), filter_string=filter_string) + return filter_needs(list(self.needs.values()), needs_config, filter_string=filter_string) def imports(self, *args): if not self.parent_need_id: diff --git a/sphinx_needs/filter_common.py b/sphinx_needs/filter_common.py index 5d68dff6c..e5e9bed04 100644 --- a/sphinx_needs/filter_common.py +++ b/sphinx_needs/filter_common.py @@ -102,6 +102,7 @@ def process_filters( :return: list of needs, which passed the filters """ + needs_config = NeedsSphinxConfig(app.config) found_needs: list[NeedsPartsInfoType] sort_key = filter_data["sort_by"] if sort_key: @@ -156,13 +157,13 @@ def process_filters( if status_filter_passed and tags_filter_passed and type_filter_passed: found_needs_by_options.append(need_info) # Get need by filter string - found_needs_by_string = filter_needs(app, all_needs_incl_parts, filter_data["filter"]) + found_needs_by_string = filter_needs(all_needs_incl_parts, needs_config, filter_data["filter"]) # Make an intersection of both lists found_needs = intersection_of_need_results(found_needs_by_options, found_needs_by_string) else: # There is no other config as the one for filter string. # So we only need this result. - found_needs = filter_needs(app, all_needs_incl_parts, filter_data["filter"]) + found_needs = filter_needs(all_needs_incl_parts, needs_config, filter_data["filter"]) else: # Provides only a copy of needs to avoid data manipulations. context = { @@ -192,7 +193,7 @@ def process_filters( found_needs = [] # Check if config allow unsafe filters - if NeedsSphinxConfig(app.config).allow_unsafe_filters: + if needs_config.allow_unsafe_filters: found_needs = found_dirty_needs else: # Just take the ids from search result and use the related, but original need @@ -203,8 +204,7 @@ def process_filters( # Store basic filter configuration and result global list. # Needed mainly for exporting the result to needs.json (if builder "needs" is used). - env = app.env - filter_list = SphinxNeedsData(env).get_or_create_filters() + filter_list = SphinxNeedsData(app.env).get_or_create_filters() found_needs_ids = [need["id_complete"] for need in found_needs] filter_list[filter_data["target_id"]] = { @@ -258,8 +258,8 @@ def intersection_of_need_results(list_a: list[T], list_b: list[T]) -> list[T]: @measure_time("filtering") def filter_needs( - app: Sphinx, needs: Iterable[V], + config: NeedsSphinxConfig, filter_string: None | str = "", current_need: NeedsInfoType | None = None, ) -> list[V]: @@ -267,14 +267,13 @@ def filter_needs( Filters given needs based on a given filter string. Returns all needs, which pass the given filter. - :param app: Sphinx application object :param needs: list of needs, which shall be filtered + :param config: NeedsSphinxConfig object :param filter_string: strings, which gets evaluated against each need :param current_need: current need, which uses the filter. :return: list of found needs """ - if not filter_string: return list(needs) @@ -286,7 +285,7 @@ def filter_needs( for filter_need in needs: try: if filter_single_need( - app, filter_need, filter_string, needs, current_need, filter_compiled=filter_compiled + filter_need, config, filter_string, needs, current_need, filter_compiled=filter_compiled ): found_needs.append(filter_need) except Exception as e: @@ -300,8 +299,8 @@ def filter_needs( @measure_time("filtering") def filter_single_need( - app: Sphinx, need: NeedsInfoType, + config: NeedsSphinxConfig, filter_string: str = "", needs: Iterable[NeedsInfoType] | None = None, current_need: NeedsInfoType | None = None, @@ -310,8 +309,8 @@ def filter_single_need( """ Checks if a single need/need_part passes a filter_string - :param app: Sphinx application object - :param current_need: + :param need: the data for a single need + :param config: NeedsSphinxConfig object :param filter_compiled: An already compiled filter_string to safe time :param need: need or need_part :param filter_string: string, which is used as input for eval() @@ -327,7 +326,7 @@ def filter_single_need( filter_context["current_need"] = need # Get needs external filter data and merge to filter_context - filter_context.update(NeedsSphinxConfig(app.config).filter_data) + filter_context.update(config.filter_data) filter_context["search"] = re.search result = False diff --git a/sphinx_needs/functions/common.py b/sphinx_needs/functions/common.py index 29638c123..69d6f4e37 100644 --- a/sphinx_needs/functions/common.py +++ b/sphinx_needs/functions/common.py @@ -11,6 +11,7 @@ from sphinx.application import Sphinx from sphinx_needs.api.exceptions import NeedsInvalidFilter +from sphinx_needs.config import NeedsSphinxConfig from sphinx_needs.data import NeedsInfoType from sphinx_needs.filter_common import filter_needs, filter_single_need from sphinx_needs.utils import logger @@ -151,7 +152,7 @@ def copy( need = needs[need_id] if filter: - result = filter_needs(app, needs.values(), filter, need) + result = filter_needs(needs.values(), NeedsSphinxConfig(app.config), filter, need) if result: need = result[0] @@ -308,6 +309,7 @@ def check_linked_values( :param one_hit: If True, only one linked need must have a positive check :return: result, if all checks are positive """ + needs_config = NeedsSphinxConfig(app.config) links = need["links"] if not isinstance(search_value, list): search_value = [search_value] @@ -316,7 +318,7 @@ def check_linked_values( need = needs[link] if filter_string: try: - if not filter_single_need(app, need, filter_string): + if not filter_single_need(need, needs_config, filter_string): continue except Exception as e: logger.warning(f"CheckLinkedValues: Filter {filter_string} not valid: Error: {e} [needs]", type="needs") @@ -417,6 +419,7 @@ def calc_sum( :return: A float number """ + needs_config = NeedsSphinxConfig(app.config) check_needs = [needs[link] for link in need["links"]] if links_only else needs.values() calculated_sum = 0.0 @@ -424,7 +427,7 @@ def calc_sum( for check_need in check_needs: if filter: try: - if not filter_single_need(app, check_need, filter): + if not filter_single_need(check_need, needs_config, filter): continue except ValueError: pass @@ -506,9 +509,10 @@ def links_from_content( raw_links.append(link[1]) if filter: + needs_config = NeedsSphinxConfig(app.config) filtered_links = [] for link in raw_links: - if link not in filtered_links and filter_single_need(app, needs[link], filter): + if link not in filtered_links and filter_single_need(needs[link], needs_config, filter): filtered_links.append(link) return filtered_links diff --git a/sphinx_needs/functions/functions.py b/sphinx_needs/functions/functions.py index ab17613bc..97c7334a2 100644 --- a/sphinx_needs/functions/functions.py +++ b/sphinx_needs/functions/functions.py @@ -16,7 +16,7 @@ from sphinx.errors import SphinxError from sphinx_needs.config import NeedsSphinxConfig -from sphinx_needs.data import SphinxNeedsData +from sphinx_needs.data import NeedsInfoType, SphinxNeedsData from sphinx_needs.debug import measure_time_func from sphinx_needs.logging import get_logger from sphinx_needs.utils import NEEDS_FUNCTIONS, match_variants # noqa: F401 @@ -25,8 +25,10 @@ unicode = str ast_boolean = ast.NameConstant -# TODO this input args should actually be of type `Need` and `List[Need]`, however `Need` is *currently* untyped. -DynamicFunction = Callable[[Sphinx, Any, Any], Union[str, int, float, List[Union[str, int, float]]]] +# TODO these functions also take optional *args and **kwargs +DynamicFunction = Callable[ + [Sphinx, NeedsInfoType, Dict[str, NeedsInfoType]], Union[str, int, float, List[Union[str, int, float]]] +] def register_func(need_function: DynamicFunction, name: Optional[str] = None) -> None: @@ -56,12 +58,12 @@ def register_func(need_function: DynamicFunction, name: Optional[str] = None) -> NEEDS_FUNCTIONS[func_name] = {"name": func_name, "function": need_function} -def execute_func(env: BuildEnvironment, need, func_string: str): - """ - Executes a given function string. +def execute_func(app: Sphinx, need: NeedsInfoType, func_string: str): + """Executes a given function string. + :param env: Sphinx environment :param need: Actual need, which contains the found function string - :param func_string: string of the found function. Without [[ ]] + :param func_string: string of the found function. Without ``[[ ]]`` :return: return value of executed function """ global NEEDS_FUNCTIONS @@ -71,7 +73,7 @@ def execute_func(env: BuildEnvironment, need, func_string: str): raise SphinxError("Unknown dynamic sphinx-needs function: {}. Found in need: {}".format(func_name, need["id"])) func = measure_time_func(NEEDS_FUNCTIONS[func_name]["function"], category="dyn_func", source="user") - func_return = func(env.app, need, SphinxNeedsData(env).get_or_create_needs(), *func_args, **func_kwargs) + func_return = func(app, need, SphinxNeedsData(app.env).get_or_create_needs(), *func_args, **func_kwargs) if not isinstance(func_return, (str, int, float, list, unicode)) and func_return: raise SphinxError( @@ -93,7 +95,7 @@ def execute_func(env: BuildEnvironment, need, func_string: str): func_pattern = re.compile(r"\[\[(.*?)\]\]") # RegEx to detect function strings -def find_and_replace_node_content(node, env: BuildEnvironment, need): +def find_and_replace_node_content(node, env: BuildEnvironment, need: NeedsInfoType): """ Search inside a given node and its children for nodes of type Text, if found, check if it contains a function string and run/replace it. @@ -125,7 +127,7 @@ def find_and_replace_node_content(node, env: BuildEnvironment, need): func_string = func_string.replace("‘", "'") func_string = func_string.replace("’", "'") - func_return = execute_func(env, need, func_string) + func_return = execute_func(env.app, need, func_string) # This should never happen, but we can not be sure. if isinstance(func_return, list): @@ -151,20 +153,25 @@ def find_and_replace_node_content(node, env: BuildEnvironment, need): return node -def resolve_dynamic_values(env: BuildEnvironment) -> None: +def resolve_dynamic_values(needs: Dict[str, NeedsInfoType], app: Sphinx) -> None: """ Resolve dynamic values inside need data. Rough workflow: - #. Parse all needs and their data for a string like ``[[ my_func(a,b,c) ]]`` + #. Parse all needs and their field values for a string like ``[[ my_func(a, b, c) ]]`` #. Extract function name and call parameters #. Execute registered function name with extracted call parameters #. Replace original string with return value - """ - data = SphinxNeedsData(env) - needs = data.get_or_create_needs() + The registered functions should take the following parameters: + + - ``app``: Sphinx application + - ``need``: Need data + - ``all_needs``: All needs of the current sphinx project + - ``*args``: optional arguments (specified in the function string) + - ``**kwargs``: optional keyword arguments (specified in the function string) + """ for need in needs.values(): for need_option in need: if need_option in ["docname", "lineno", "content", "content_node", "content_id"]: @@ -174,7 +181,7 @@ def resolve_dynamic_values(env: BuildEnvironment) -> None: func_call = True while func_call: try: - func_call, func_return = _detect_and_execute(need[need_option], need, env) + func_call, func_return = _detect_and_execute(need[need_option], need, app) except FunctionParsingException: raise SphinxError( "Function definition of {option} in file {file}:{line} has " @@ -198,7 +205,7 @@ def resolve_dynamic_values(env: BuildEnvironment) -> None: new_values = [] for element in need[need_option]: try: - func_call, func_return = _detect_and_execute(element, need, env) + func_call, func_return = _detect_and_execute(element, need, app) except FunctionParsingException: raise SphinxError( "Function definition of {option} in file {file}:{line} has " @@ -223,29 +230,31 @@ def resolve_dynamic_values(env: BuildEnvironment) -> None: need[need_option] = new_values -def resolve_variants_options(env: BuildEnvironment) -> None: +def resolve_variants_options( + needs: Dict[str, NeedsInfoType], needs_config: NeedsSphinxConfig, tags: Dict[str, bool] +) -> None: """ Resolve variants options inside need data. + These are fields specified by the user, + that have string values with a special markup syntax like ``var_a:open``. + These need to be resolved to the actual value. + Rough workflow: - #. Parse all needs and their data for variant handling + #. Parse all needs and their fields for variant handling #. Replace original string with return value """ - data = SphinxNeedsData(env) - needs_config = NeedsSphinxConfig(env.config) variants_options = needs_config.variant_options if not variants_options: return - needs = data.get_or_create_needs() for need in needs.values(): # Data to use as filter context. need_context: Dict[str, Any] = {**need} need_context.update(**needs_config.filter_data) # Add needs_filter_data to filter context - _sphinx_tags = env.app.builder.tags.tags # Get sphinx tags - need_context.update(**_sphinx_tags) # Add sphinx tags to filter context + need_context.update(**tags) # Add sphinx tags to filter context for var_option in variants_options: if var_option in need and need[var_option] not in (None, "", []): @@ -279,14 +288,15 @@ def check_and_get_content(content: str, need, env: BuildEnvironment) -> str: return content func_call = func_match.group(1) # Extract function call - func_return = execute_func(env, need, func_call) # Execute function call and get return value + func_return = execute_func(env.app, need, func_call) # Execute function call and get return value # Replace the function_call with the calculated value content = content.replace(f"[[{func_call}]]", func_return) return content -def _detect_and_execute(content, need, env): +def _detect_and_execute(content: Any, need: NeedsInfoType, app: Sphinx): + """Detects if given content is a function call and executes it.""" try: content = str(content) except UnicodeEncodeError: @@ -297,7 +307,7 @@ def _detect_and_execute(content, need, env): return None, None func_call = func_match.group(1) # Extract function call - func_return = execute_func(env, need, func_call) # Execute function call and get return value + func_return = execute_func(app, need, func_call) # Execute function call and get return value return func_call, func_return diff --git a/sphinx_needs/need_constraints.py b/sphinx_needs/need_constraints.py index 52db77436..77db7d03b 100644 --- a/sphinx_needs/need_constraints.py +++ b/sphinx_needs/need_constraints.py @@ -1,18 +1,17 @@ from typing import Dict import jinja2 -from sphinx.application import Sphinx from sphinx_needs.api.exceptions import NeedsConstraintFailed, NeedsConstraintNotAllowed from sphinx_needs.config import NeedsSphinxConfig -from sphinx_needs.data import SphinxNeedsData +from sphinx_needs.data import NeedsInfoType from sphinx_needs.filter_common import filter_single_need from sphinx_needs.logging import get_logger logger = get_logger(__name__) -def process_constraints(app: Sphinx) -> None: +def process_constraints(needs: Dict[str, NeedsInfoType], config: NeedsSphinxConfig) -> None: """Analyse constraints of all needs, and set corresponding fields on the need data item: ``constraints_passed`` and ``constraints_results``. @@ -20,10 +19,7 @@ def process_constraints(app: Sphinx) -> None: The ``style`` field may also be changed, if a constraint fails (depending on the config value ``constraint_failed_options``) """ - env = app.env - needs_config = NeedsSphinxConfig(env.config) - config_constraints = needs_config.constraints - needs = SphinxNeedsData(env).get_or_create_needs() + config_constraints = config.constraints error_templates_cache: Dict[str, jinja2.Template] = {} @@ -50,7 +46,7 @@ def process_constraints(app: Sphinx) -> None: continue # compile constraint and check if need fulfils it - constraint_passed = filter_single_need(app, need, cmd) + constraint_passed = filter_single_need(need, config, cmd) if constraint_passed: need["constraints_results"].setdefault(constraint, {})[name] = True @@ -68,11 +64,11 @@ def process_constraints(app: Sphinx) -> None: f"'severity' key not set for constraint {constraint!r} in config 'needs_constraints'" ) severity = executable_constraints["severity"] - if severity not in needs_config.constraint_failed_options: + if severity not in config.constraint_failed_options: raise NeedsConstraintFailed( f"Severity {severity!r} not set in config 'needs_constraint_failed_options'" ) - failed_options = needs_config.constraint_failed_options[severity] + failed_options = config.constraint_failed_options[severity] # log/except if needed if "warn" in failed_options.get("on_fail", []): diff --git a/sphinx_needs/roles/need_count.py b/sphinx_needs/roles/need_count.py index cc5336580..cfe996c7b 100644 --- a/sphinx_needs/roles/need_count.py +++ b/sphinx_needs/roles/need_count.py @@ -10,6 +10,7 @@ from sphinx.application import Sphinx from sphinx_needs.api.exceptions import NeedsInvalidFilter +from sphinx_needs.config import NeedsSphinxConfig from sphinx_needs.data import SphinxNeedsData from sphinx_needs.filter_common import filter_needs, prepare_need_list from sphinx_needs.logging import get_logger @@ -24,21 +25,20 @@ class NeedCount(nodes.Inline, nodes.Element): # type: ignore def process_need_count( app: Sphinx, doctree: nodes.document, _fromdocname: str, found_nodes: List[nodes.Element] ) -> None: - env = app.env - # for node_need_count in doctree.findall(NeedCount): + needs_config = NeedsSphinxConfig(app.config) for node_need_count in found_nodes: - all_needs = list(SphinxNeedsData(env).get_or_create_needs().values()) + all_needs = list(SphinxNeedsData(app.env).get_or_create_needs().values()) filter = node_need_count["reftarget"] if filter: filters = filter.split(" ? ") if len(filters) == 1: need_list = prepare_need_list(all_needs) # adds parts to need_list - amount = str(len(filter_needs(app, need_list, filters[0]))) + amount = str(len(filter_needs(need_list, needs_config, filters[0]))) elif len(filters) == 2: need_list = prepare_need_list(all_needs) # adds parts to need_list - amount_1 = len(filter_needs(app, need_list, filters[0])) - amount_2 = len(filter_needs(app, need_list, filters[1])) + amount_1 = len(filter_needs(need_list, needs_config, filters[0])) + amount_2 = len(filter_needs(need_list, needs_config, filters[1])) amount = f"{amount_1 / amount_2 * 100:2.1f}" elif len(filters) > 2: raise NeedsInvalidFilter( diff --git a/sphinx_needs/warnings.py b/sphinx_needs/warnings.py index 4e237cb60..8754551dc 100644 --- a/sphinx_needs/warnings.py +++ b/sphinx_needs/warnings.py @@ -32,8 +32,9 @@ def process_warnings(app: Sphinx, exception: Optional[Exception]) -> None: return env = app.env + needs = SphinxNeedsData(env).get_or_create_needs() # If no needs were defined, we do not need to do anything - if not hasattr(env, "needs_all_needs"): + if not needs: return # Check if warnings already got executed. @@ -44,15 +45,14 @@ def process_warnings(app: Sphinx, exception: Optional[Exception]) -> None: env.needs_warnings_executed = True # type: ignore[attr-defined] - needs = SphinxNeedsData(env).get_or_create_needs() - # Exclude external needs for warnings check checked_needs: Dict[str, NeedsInfoType] = {} for need_id, need in needs.items(): if not need["is_external"]: checked_needs[need_id] = need - warnings_always_warn = NeedsSphinxConfig(app.config).warnings_always_warn + needs_config = NeedsSphinxConfig(app.config) + warnings_always_warn = needs_config.warnings_always_warn with logging.pending_logging(): logger.info("\nChecking sphinx-needs warnings") @@ -60,7 +60,7 @@ def process_warnings(app: Sphinx, exception: Optional[Exception]) -> None: for warning_name, warning_filter in NEEDS_CONFIG.warnings.items(): if isinstance(warning_filter, str): # filter string used - result = filter_needs(app, checked_needs.values(), warning_filter) + result = filter_needs(checked_needs.values(), needs_config, warning_filter) elif callable(warning_filter): # custom defined filter code used from conf.py result = []