File: external_needs.py

package info (click to toggle)
sphinx-needs 5.1.0%2Bdfsg-6
  • links: PTS, VCS
  • area: main
  • in suites: sid
  • size: 12,108 kB
  • sloc: python: 21,148; javascript: 187; makefile: 95; sh: 29; xml: 10
file content (221 lines) | stat: -rw-r--r-- 8,466 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
from __future__ import annotations

import json
import os
from functools import lru_cache

import requests
from jinja2 import Environment, Template
from requests_file import FileAdapter
from sphinx.application import Sphinx
from sphinx.environment import BuildEnvironment

from sphinx_needs.api import InvalidNeedException, add_external_need, del_need
from sphinx_needs.config import NeedsSphinxConfig
from sphinx_needs.data import NeedsCoreFields, SphinxNeedsData
from sphinx_needs.logging import get_logger, log_warning
from sphinx_needs.utils import clean_log, import_prefix_link_edit

log = get_logger(__name__)


@lru_cache(maxsize=20)
def get_target_template(target_url: str) -> Template:
    """
    Provides template for target_link style
    Can be cached, as the template is always the same for a given target_url
    """
    mem_template = Environment().from_string(target_url)
    return mem_template


def load_external_needs(
    app: Sphinx, env: BuildEnvironment, _docnames: list[str]
) -> None:
    """Load needs from configured external sources."""
    needs_config = NeedsSphinxConfig(app.config)
    for source in needs_config.external_needs:
        if source["base_url"].endswith("/"):
            source["base_url"] = source["base_url"][:-1]

        target_url = source.get("target_url", "")

        if source.get("json_url", False) and source.get("json_path", False):
            raise NeedsExternalException(
                clean_log(
                    "json_path and json_url are both configured, but only one is allowed.\n"
                    f"json_path: {source['json_path']}\n"
                    f"json_url: {source['json_url']}."
                )
            )
        elif not (source.get("json_url", False) or source.get("json_path", False)):
            raise NeedsExternalException(
                "json_path or json_url must be configured to use external_needs."
            )

        if source.get("json_url", False):
            log.info(
                clean_log(f"Loading external needs from url {source['json_url']}.")
            )
            s = requests.Session()
            s.mount("file://", FileAdapter())
            try:
                response = s.get(source["json_url"])
                needs_json = (
                    response.json()
                )  # The downloaded file MUST be json. Everything else we do not handle!
            except Exception as e:
                raise NeedsExternalException(
                    clean_log(
                        "Getting {} didn't work. Reason: {}".format(
                            source["json_url"], e
                        )
                    )
                )

        if source.get("json_path", False):
            if os.path.isabs(source["json_path"]):
                json_path = source["json_path"]
            else:
                json_path = os.path.join(app.confdir, source["json_path"])

            if not os.path.exists(json_path):
                raise NeedsExternalException(
                    f"Given json_path {json_path} does not exist."
                )

            with open(json_path) as json_file:
                needs_json = json.load(json_file)

        version = source.get("version", needs_json.get("current_version"))
        if not version:
            raise NeedsExternalException(
                'No version defined in "needs_external_needs" or by "current_version" from loaded json file'
            )

        try:
            data = needs_json["versions"][version]
            needs = data["needs"]
        except KeyError:
            uri = source.get("json_url", source.get("json_path", "unknown"))
            raise NeedsExternalException(
                clean_log(
                    f"Version {version} not found in json file from {uri}: {list(needs_json.get('versions'))}"
                )
            )

        log.debug(f"Loading {len(needs)} needs.")

        defaults = (
            {
                name: value["default"]
                for name, value in schema["properties"].items()
                if "default" in value
            }
            if (schema := data.get("needs_schema"))
            else {}
        )

        id_prefix = source.get("id_prefix", "").upper()
        import_prefix_link_edit(needs, id_prefix, needs_config.extra_links)

        # all known need fields in the project
        known_keys = {
            "full_title",  # legacy
            *NeedsCoreFields,
            *(x["option"] for x in needs_config.extra_links),
            *(x["option"] + "_back" for x in needs_config.extra_links),
            *needs_config.extra_options,
        }
        # all keys that should not be imported from external needs
        omitted_keys = {
            "full_title",  # legacy
            *(k for k, v in NeedsCoreFields.items() if v.get("exclude_external")),
            *(x["option"] + "_back" for x in needs_config.extra_links),
        }

        # collect keys for warning logs, so that we only log one warning per key
        unknown_keys: set[str] = set()
        non_string_extra_keys: set[str] = set()

        for need in needs.values():
            need_params = {**defaults, **need}

            if "description" in need_params and not need_params.get("content"):
                # legacy versions of sphinx-needs changed "description" to "content" when outputting to json
                need_params["content"] = need_params["description"]
                del need_params["description"]

            # Remove unknown options, as they may be defined in source system, but not in this sphinx project
            for option in list(need_params):
                if option not in known_keys:
                    unknown_keys.add(option)
                    del need_params[option]
                elif option in omitted_keys:
                    del need_params[option]
                if option in needs_config.extra_options and not isinstance(
                    need_params[option], str
                ):
                    non_string_extra_keys.add(option)

            # These keys need to be different for add_need() api call.
            need_params["need_type"] = need_params.pop("type", "")

            # Replace id, to get unique ids
            need_params["id"] = id_prefix + need["id"]

            need_params["external_css"] = source.get("css_class")

            if target_url:
                # render jinja content
                mem_template = get_target_template(target_url)
                cal_target_url = mem_template.render(**{"need": need})
                need_params["external_url"] = f"{source['base_url']}/{cal_target_url}"
            else:
                need_params["external_url"] = (
                    f"{source['base_url']}/{need.get('docname', '__error__')}.html#{need['id']}"
                )

            # check if external needs already exist
            ext_need_id = need_params["id"]

            need = SphinxNeedsData(env).get_needs_mutable().get(ext_need_id)

            if (
                need is not None
                and need["is_external"]
                and source["base_url"] in need["external_url"]
            ):
                # delete the already existing external need from api need
                del_need(app, ext_need_id)

            try:
                add_external_need(app, **need_params)
            except InvalidNeedException as err:
                location = source.get("json_url", "") or source.get("json_path", "")
                log_warning(
                    log,
                    f"External need {ext_need_id!r} in {location!r} could not be added: {err.message}",
                    "load_external_need",
                    location=None,
                )

        source_str = source.get("json_url", "") or source.get("json_path", "")
        if unknown_keys:
            log_warning(
                log,
                f"Unknown keys in external need source {source_str!r}: {sorted(unknown_keys)!r}",
                "unknown_external_keys",
                location=None,
            )
        if non_string_extra_keys:
            log_warning(
                log,
                f"Non-string values in extra options of external need source {source_str!r}: {sorted(non_string_extra_keys)!r}",
                "mistyped_external_values",
                location=None,
            )


class NeedsExternalException(BaseException):
    pass