Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

🔧 Make NEEDS_CONFIG type safe #1001

Merged
merged 3 commits into from
Aug 28, 2023
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 10 additions & 12 deletions sphinx_needs/api/configuration.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,14 +3,15 @@

All functions here are available under ``sphinxcontrib.api``. So do not import this module directly.
"""
from typing import Any, Callable, List, Optional
from typing import Callable, List, Optional

from docutils.parsers.rst import directives
from sphinx.application import Sphinx
from sphinx.util.logging import SphinxLoggerAdapter

from sphinx_needs.api.exceptions import NeedsApiConfigException, NeedsApiConfigWarning
from sphinx_needs.config import NEEDS_CONFIG, NeedsSphinxConfig
from sphinx_needs.data import NeedsInfoType
from sphinx_needs.functions import register_func
from sphinx_needs.functions.functions import DynamicFunction

Expand Down Expand Up @@ -84,13 +85,9 @@ def add_extra_option(app: Sphinx, name: str) -> None:
:param name: Name as string of the extra option
:return: None
"""

extra_options = NEEDS_CONFIG.create_or_get("extra_options", dict)

if name in extra_options:
if name in NEEDS_CONFIG.extra_options:
raise NeedsApiConfigWarning(f"Option {name} already registered.")

NEEDS_CONFIG.add("extra_options", {name: directives.unchanged}, dict, append=True)
NEEDS_CONFIG.extra_options[name] = directives.unchanged


def add_dynamic_function(app: Sphinx, function: DynamicFunction, name: Optional[str] = None) -> None:
Expand Down Expand Up @@ -121,7 +118,7 @@ def my_function(app, need, needs, *args, **kwargs):


# 'Need' is untyped, so we temporarily use 'Any' here
WarningCheck = Callable[[Any, SphinxLoggerAdapter], bool]
WarningCheck = Callable[[NeedsInfoType, SphinxLoggerAdapter], bool]


def add_warning(
Expand All @@ -138,8 +135,6 @@ def add_warning(
:param filter_string: filter_string to use for the warning
:return: None
"""
warnings_option = NEEDS_CONFIG.create_or_get("warnings", dict)

if function is None and filter_string is None:
raise NeedsApiConfigException("Function or filter_string must be given for add_warning_func")

Expand All @@ -150,7 +145,10 @@ def add_warning(

warning_check = function or filter_string

if name in warnings_option:
if warning_check is None:
raise NeedsApiConfigException("either function or filter_string must be given")

if name in NEEDS_CONFIG.warnings:
raise NeedsApiConfigException(f"Warning {name} already registered.")

NEEDS_CONFIG.add("warnings", {name: warning_check}, dict, append=True)
NEEDS_CONFIG.warnings[name] = warning_check
2 changes: 1 addition & 1 deletion sphinx_needs/api/need.py
Original file line number Diff line number Diff line change
Expand Up @@ -346,7 +346,7 @@ def run():
"is_modified": False, # needed by needextend
"modifications": 0, # needed by needextend
}
needs_extra_option_names = NEEDS_CONFIG.get("extra_options").keys()
needs_extra_option_names = list(NEEDS_CONFIG.extra_options)
_merge_extra_options(needs_info, kwargs, needs_extra_option_names)

needs_global_options = needs_config.global_options
Expand Down
63 changes: 30 additions & 33 deletions sphinx_needs/config.py
Original file line number Diff line number Diff line change
@@ -1,13 +1,18 @@
from __future__ import annotations

from dataclasses import MISSING, dataclass, field, fields
from typing import Any, Callable
from typing import TYPE_CHECKING, Any, Callable

from sphinx.application import Sphinx
from sphinx.config import Config as _SphinxConfig

from sphinx_needs.defaults import DEFAULT_DIAGRAM_TEMPLATE, NEEDS_TABLES_CLASSES

if TYPE_CHECKING:
from sphinx.util.logging import SphinxLoggerAdapter

from sphinx_needs.data import NeedsInfoType


class Config:
"""
Expand All @@ -20,38 +25,30 @@ class Config:
"""

def __init__(self) -> None:
self.configs: dict[str, Any] = {}

def add(
self, name: str, value: Any, option_type: type = str, append: bool = False, overwrite: bool = False
) -> None:
if name not in self.configs:
self.configs[name] = option_type()
elif not isinstance(self.configs[name], option_type):
raise Exception(
f"Type of needs config option {name} is {type(self.configs[name])}," f"but {option_type} is given"
)

if not append:
self.configs[name] = value
else:
if isinstance(self.configs[name], dict):
self.configs[name] = {**self.configs[name], **value}
else:
self.configs[name] += value

def create(self, name: str, option_type: Callable[[], Any] = str, overwrite: bool = False) -> None:
if name in self.configs and not overwrite:
raise Exception(f"option {name} exists.")
self.configs[name] = option_type()

def create_or_get(self, name: str, option_type: Callable[[], Any] = str) -> Any:
if name not in self.configs:
self.configs[name] = option_type()
return self.configs[name]

def get(self, name: str) -> Any:
return self.configs[name]
self._extra_options: dict[str, Callable[[str], Any]] = {}
self._warnings: dict[str, str | Callable[[NeedsInfoType, SphinxLoggerAdapter], bool]] = {}

def clear(self) -> None:
self._extra_options = {}
self._warnings = {}

@property
def extra_options(self) -> dict[str, Callable[[str], Any]]:
"""Options that are dynamically added to `NeedDirective` & `NeedserviceDirective`,
after the config is initialized.

These fields are also added to the each needs data item.

:returns: Mapping of name to validation function
"""
return self._extra_options

@property
def warnings(self) -> dict[str, str | Callable[[NeedsInfoType, SphinxLoggerAdapter], bool]]:
"""Warning handlers that are added by the user,
then called at the end of the build.
"""
return self._warnings


NEEDS_CONFIG = Config()
Expand Down
2 changes: 1 addition & 1 deletion sphinx_needs/directives/need.py
Original file line number Diff line number Diff line change
Expand Up @@ -122,7 +122,7 @@ def run(self) -> Sequence[nodes.Node]:
for extra_link in self.needs_config.extra_links:
need_extra_options[extra_link["option"]] = self.options.get(extra_link["option"], "")

for extra_option in NEEDS_CONFIG.get("extra_options").keys():
for extra_option in NEEDS_CONFIG.extra_options:
need_extra_options[extra_option] = self.options.get(extra_option, "")

need_nodes = add_need(
Expand Down
2 changes: 1 addition & 1 deletion sphinx_needs/directives/needimport.py
Original file line number Diff line number Diff line change
Expand Up @@ -196,7 +196,7 @@ def run(self) -> Sequence[nodes.Node]:
need["content"] = need["description"]
# Remove unknown options, as they may be defined in source system, but not in this sphinx project
extra_link_keys = [x["option"] for x in extra_links]
extra_option_keys = list(NEEDS_CONFIG.get("extra_options").keys())
extra_option_keys = list(NEEDS_CONFIG.extra_options)
default_options = [
"title",
"status",
Expand Down
20 changes: 8 additions & 12 deletions sphinx_needs/needs.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,6 @@
from sphinx.errors import SphinxError

import sphinx_needs.debug as debug # Need to set global var in it for timeing measurements
from sphinx_needs.api.configuration import add_extra_option
from sphinx_needs.builder import (
NeedsBuilder,
NeedumlsBuilder,
Expand Down Expand Up @@ -243,8 +242,7 @@ def setup(app: Sphinx) -> Dict[str, Any]:

# Be sure Sphinx-Needs config gets erased before any events or external API calls get executed.
# So never but this inside an event.
NEEDS_CONFIG.create("extra_options", dict, overwrite=True)
NEEDS_CONFIG.create("warnings", dict, overwrite=True)
NEEDS_CONFIG.clear()

return {
"version": VERSION,
Expand Down Expand Up @@ -308,12 +306,11 @@ def load_config(app: Sphinx, *_args) -> None:
"Sphinx-Needs 0.7.2 is list. Please see docs for details."
)

existing_extra_options = NEEDS_CONFIG.get("extra_options")
extra_options = NEEDS_CONFIG.extra_options
for option in needs_config.extra_options:
if option in existing_extra_options:
if option in extra_options:
log.warning(f'extra_option "{option}" already registered. [needs]', type="needs")
NEEDS_CONFIG.add("extra_options", {option: directives.unchanged}, dict, True)
extra_options = NEEDS_CONFIG.get("extra_options")
NEEDS_CONFIG.extra_options[option] = directives.unchanged

# Get extra links and create a dictionary of needed options.
extra_links_raw = needs_config.extra_links
Expand Down Expand Up @@ -384,10 +381,9 @@ def load_config(app: Sphinx, *_args) -> None:
# Register requested types of needs
app.add_directive(t["directive"], NeedDirective)

existing_warnings = NEEDS_CONFIG.get("warnings")
for name, check in needs_config.warnings.items():
if name not in existing_warnings:
NEEDS_CONFIG.add("warnings", {name: check}, dict, append=True)
if name not in NEEDS_CONFIG.warnings:
NEEDS_CONFIG.warnings[name] = check
else:
log.warning(f'{name} for "warnings" is already registered. [needs]', type="needs")

Expand Down Expand Up @@ -437,8 +433,8 @@ def prepare_env(app: Sphinx, env: BuildEnvironment, _docname: str) -> None:
# Own extra options
for option in ["hidden", "duration", "completion", "has_dead_links", "has_forbidden_dead_links", "constraints"]:
# Check if not already set by user
if option not in NEEDS_CONFIG.get("extra_options"):
add_extra_option(app, option)
if option not in NEEDS_CONFIG.extra_options:
NEEDS_CONFIG.extra_options[option] = directives.unchanged

# The default link name. Must exist in all configurations. Therefore we set it here
# for the user.
Expand Down
5 changes: 2 additions & 3 deletions sphinx_needs/services/manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,10 +26,9 @@ def register(self, name: str, clazz, **kwargs) -> None:

# Register options from service class
for option in clazz.options:
extra_option_names = NEEDS_CONFIG.get("extra_options").keys()
if option not in extra_option_names:
if option not in NEEDS_CONFIG.extra_options:
self.log.debug(f'Register option "{option}" for service "{name}"')
NEEDS_CONFIG.add("extra_options", {option: directives.unchanged}, dict, append=True)
NEEDS_CONFIG.extra_options[option] = directives.unchanged
# Register new option directly in Service directive, as its class options got already
# calculated
NeedserviceDirective.option_spec[option] = directives.unchanged
Expand Down
10 changes: 4 additions & 6 deletions sphinx_needs/warnings.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,13 +2,13 @@
Cares about handling and execution warnings.

"""
from typing import Optional
from typing import Dict, Optional

from sphinx.application import Sphinx
from sphinx.util import logging

from sphinx_needs.config import NEEDS_CONFIG, NeedsSphinxConfig
from sphinx_needs.data import SphinxNeedsData
from sphinx_needs.data import NeedsInfoType, SphinxNeedsData
from sphinx_needs.filter_common import filter_needs
from sphinx_needs.logging import get_logger
from sphinx_needs.utils import unwrap
Expand Down Expand Up @@ -48,19 +48,17 @@ def process_warnings(app: Sphinx, exception: Optional[Exception]) -> None:
needs = SphinxNeedsData(env).get_or_create_needs()

# Exclude external needs for warnings check
checked_needs = {}
checked_needs: Dict[str, NeedsInfoType] = {}
for need_id, need in needs.items():
if not need["is_external"]:
checked_needs[need_id] = need

warnings = NEEDS_CONFIG.get("warnings")

warnings_always_warn = NeedsSphinxConfig(app.config).warnings_always_warn

with logging.pending_logging():
logger.info("\nChecking sphinx-needs warnings")
warning_raised = False
for warning_name, warning_filter in warnings.items():
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)
Expand Down