Split `sphinx.ext.intersphinx._load`

Co-authored-by: Adam Turner <9087854+aa-turner@users.noreply.github.com>

.ruff.toml

@@ -505,6 +505,7 @@ exclude = [
     "sphinx/ext/autosectionlabel.py",
     "sphinx/ext/intersphinx/__init__.py",
     "sphinx/ext/intersphinx/_cli.py",
+    "sphinx/ext/intersphinx/_load.py",
     "sphinx/ext/duration.py",
     "sphinx/ext/imgconverter.py",
     "sphinx/ext/imgmath.py",

pyproject.toml

@@ -263,7 +263,7 @@ module = [
     "sphinx.ext.doctest",
     "sphinx.ext.graphviz",
     "sphinx.ext.inheritance_diagram",
-    "sphinx.ext.intersphinx",
+    "sphinx.ext.intersphinx._load",
     "sphinx.ext.napoleon.docstring",
     "sphinx.highlighting",
     "sphinx.jinja2glue",

sphinx/ext/intersphinx/__init__.py

@@ -36,241 +36,44 @@ __all__ = (
     'inspect_main',
 )
 
-import concurrent.futures
-import functools
 import posixpath
 import re
-import time
-from os import path
 from typing import TYPE_CHECKING, cast
-from urllib.parse import urlsplit, urlunsplit
 
 from docutils import nodes
 from docutils.utils import relative_path
 
 import sphinx
 from sphinx.addnodes import pending_xref
-from sphinx.builders.html import INVENTORY_FILENAME
 from sphinx.deprecation import _deprecation_warning
 from sphinx.errors import ExtensionError
 from sphinx.ext.intersphinx._cli import inspect_main
+from sphinx.ext.intersphinx._load import (
+    fetch_inventory,
+    fetch_inventory_group,
+    load_mappings,
+    normalize_intersphinx_mapping,
+)
 from sphinx.ext.intersphinx._shared import LOGGER as logger
 from sphinx.ext.intersphinx._shared import InventoryAdapter
 from sphinx.locale import _, __
 from sphinx.transforms.post_transforms import ReferencesResolver
-from sphinx.util import requests
 from sphinx.util.docutils import CustomReSTDispatcher, SphinxRole
-from sphinx.util.inventory import InventoryFile
 
 if TYPE_CHECKING:
     from collections.abc import Iterable
     from types import ModuleType
-    from typing import IO, Any
+    from typing import Any
 
     from docutils.nodes import Node, TextElement, system_message
     from docutils.utils import Reporter
 
     from sphinx.application import Sphinx
-    from sphinx.config import Config
     from sphinx.domains import Domain
     from sphinx.environment import BuildEnvironment
-    from sphinx.ext.intersphinx._shared import InventoryCacheEntry
     from sphinx.util.typing import ExtensionMetadata, Inventory, InventoryItem, RoleFunction
 
 
-def _strip_basic_auth(url: str) -> str:
-    """Returns *url* with basic auth credentials removed. Also returns the
-    basic auth username and password if they're present in *url*.
-
-    E.g.: https://user:pass@example.com => https://example.com
-
-    *url* need not include basic auth credentials.
-
-    :param url: url which may or may not contain basic auth credentials
-    :type url: ``str``
-
-    :return: *url* with any basic auth creds removed
-    :rtype: ``str``
-    """
-    frags = list(urlsplit(url))
-    # swap out "user[:pass]@hostname" for "hostname"
-    if '@' in frags[1]:
-        frags[1] = frags[1].split('@')[1]
-    return urlunsplit(frags)
-
-
-def _read_from_url(url: str, *, config: Config) -> IO:
-    """Reads data from *url* with an HTTP *GET*.
-
-    This function supports fetching from resources which use basic HTTP auth as
-    laid out by RFC1738 § 3.1. See § 5 for grammar definitions for URLs.
-
-    .. seealso:
-
-       https://www.ietf.org/rfc/rfc1738.txt
-
-    :param url: URL of an HTTP resource
-    :type url: ``str``
-
-    :return: data read from resource described by *url*
-    :rtype: ``file``-like object
-    """
-    r = requests.get(url, stream=True, timeout=config.intersphinx_timeout,
-                     _user_agent=config.user_agent,
-                     _tls_info=(config.tls_verify, config.tls_cacerts))
-    r.raise_for_status()
-    r.raw.url = r.url
-    # decode content-body based on the header.
-    # ref: https://github.com/psf/requests/issues/2155
-    r.raw.read = functools.partial(r.raw.read, decode_content=True)
-    return r.raw
-
-
-def _get_safe_url(url: str) -> str:
-    """Gets version of *url* with basic auth passwords obscured. This function
-    returns results suitable for printing and logging.
-
-    E.g.: https://user:12345@example.com => https://user@example.com
-
-    :param url: a url
-    :type url: ``str``
-
-    :return: *url* with password removed
-    :rtype: ``str``
-    """
-    parts = urlsplit(url)
-    if parts.username is None:
-        return url
-    else:
-        frags = list(parts)
-        if parts.port:
-            frags[1] = f'{parts.username}@{parts.hostname}:{parts.port}'
-        else:
-            frags[1] = f'{parts.username}@{parts.hostname}'
-
-        return urlunsplit(frags)
-
-
-def fetch_inventory(app: Sphinx, uri: str, inv: str) -> Inventory:
-    """Fetch, parse and return an intersphinx inventory file."""
-    # both *uri* (base URI of the links to generate) and *inv* (actual
-    # location of the inventory file) can be local or remote URIs
-    if '://' in uri:
-        # case: inv URI points to remote resource; strip any existing auth
-        uri = _strip_basic_auth(uri)
-    try:
-        if '://' in inv:
-            f = _read_from_url(inv, config=app.config)
-        else:
-            f = open(path.join(app.srcdir, inv), 'rb')  # NoQA: SIM115
-    except Exception as err:
-        err.args = ('intersphinx inventory %r not fetchable due to %s: %s',
-                    inv, err.__class__, str(err))
-        raise
-    try:
-        if hasattr(f, 'url'):
-            newinv = f.url
-            if inv != newinv:
-                logger.info(__('intersphinx inventory has moved: %s -> %s'), inv, newinv)
-
-                if uri in (inv, path.dirname(inv), path.dirname(inv) + '/'):
-                    uri = path.dirname(newinv)
-        with f:
-            try:
-                invdata = InventoryFile.load(f, uri, posixpath.join)
-            except ValueError as exc:
-                raise ValueError('unknown or unsupported inventory version: %r' % exc) from exc
-    except Exception as err:
-        err.args = ('intersphinx inventory %r not readable due to %s: %s',
-                    inv, err.__class__.__name__, str(err))
-        raise
-    else:
-        return invdata
-
-
-def fetch_inventory_group(
-    name: str | None,
-    uri: str,
-    invs: tuple[str | None, ...],
-    cache: dict[str, InventoryCacheEntry],
-    app: Sphinx,
-    now: int,
-) -> bool:
-    cache_time = now - app.config.intersphinx_cache_limit * 86400
-    failures = []
-    try:
-        for inv in invs:
-            if not inv:
-                inv = posixpath.join(uri, INVENTORY_FILENAME)
-            # decide whether the inventory must be read: always read local
-            # files; remote ones only if the cache time is expired
-            if '://' not in inv or uri not in cache or cache[uri][1] < cache_time:
-                safe_inv_url = _get_safe_url(inv)
-                logger.info(__('loading intersphinx inventory from %s...'), safe_inv_url)
-                try:
-                    invdata = fetch_inventory(app, uri, inv)
-                except Exception as err:
-                    failures.append(err.args)
-                    continue
-                if invdata:
-                    cache[uri] = name, now, invdata
-                    return True
-        return False
-    finally:
-        if failures == []:
-            pass
-        elif len(failures) < len(invs):
-            logger.info(__("encountered some issues with some of the inventories,"
-                           " but they had working alternatives:"))
-            for fail in failures:
-                logger.info(*fail)
-        else:
-            issues = '\n'.join(f[0] % f[1:] for f in failures)
-            logger.warning(__("failed to reach any of the inventories "
-                              "with the following issues:") + "\n" + issues)
-
-
-def load_mappings(app: Sphinx) -> None:
-    """Load all intersphinx mappings into the environment."""
-    now = int(time.time())
-    inventories = InventoryAdapter(app.builder.env)
-    intersphinx_cache: dict[str, InventoryCacheEntry] = inventories.cache
-
-    with concurrent.futures.ThreadPoolExecutor() as pool:
-        futures = []
-        name: str | None
-        uri: str
-        invs: tuple[str | None, ...]
-        for name, (uri, invs) in app.config.intersphinx_mapping.values():
-            futures.append(pool.submit(
-                fetch_inventory_group, name, uri, invs, intersphinx_cache, app, now,
-            ))
-        updated = [f.result() for f in concurrent.futures.as_completed(futures)]
-
-    if any(updated):
-        inventories.clear()
-
-        # Duplicate values in different inventories will shadow each
-        # other; which one will override which can vary between builds
-        # since they are specified using an unordered dict.  To make
-        # it more consistent, we sort the named inventories and then
-        # add the unnamed inventories last.  This means that the
-        # unnamed inventories will shadow the named ones but the named
-        # ones can still be accessed when the name is specified.
-        named_vals = []
-        unnamed_vals = []
-        for name, _expiry, invdata in intersphinx_cache.values():
-            if name:
-                named_vals.append((name, invdata))
-            else:
-                unnamed_vals.append((name, invdata))
-        for name, invdata in sorted(named_vals) + unnamed_vals:
-            if name:
-                inventories.named_inventory[name] = invdata
-            for type, objects in invdata.items():
-                inventories.main_inventory.setdefault(type, {}).update(objects)
-
-
 def _create_element_from_result(domain: Domain, inv_name: str | None,
                                 data: InventoryItem,
                                 node: pending_xref, contnode: TextElement) -> nodes.reference:
@@ -744,39 +547,6 @@ def install_dispatcher(app: Sphinx, docname: str, source: list[str]) -> None:
     dispatcher.enable()
 
 
-def normalize_intersphinx_mapping(app: Sphinx, config: Config) -> None:
-    for key, value in config.intersphinx_mapping.copy().items():
-        try:
-            if isinstance(value, (list, tuple)):
-                # new format
-                name, (uri, inv) = key, value
-                if not isinstance(name, str):
-                    logger.warning(__('intersphinx identifier %r is not string. Ignored'),
-                                   name)
-                    config.intersphinx_mapping.pop(key)
-                    continue
-            else:
-                # old format, no name
-                # xref RemovedInSphinx80Warning
-                name, uri, inv = None, key, value
-                msg = (
-                    "The pre-Sphinx 1.0 'intersphinx_mapping' format is "
-                    "deprecated and will be removed in Sphinx 8. Update to the "
-                    "current format as described in the documentation. "
-                    f"Hint: \"intersphinx_mapping = {{'<name>': {(uri, inv)!r}}}\"."
-                    "https://www.sphinx-doc.org/en/master/usage/extensions/intersphinx.html#confval-intersphinx_mapping"  # NoQA: E501
-                )
-                logger.warning(msg)
-
-            if not isinstance(inv, tuple):
-                config.intersphinx_mapping[key] = (name, (uri, (inv,)))
-            else:
-                config.intersphinx_mapping[key] = (name, (uri, inv))
-        except Exception as exc:
-            logger.warning(__('Failed to read intersphinx_mapping[%s], ignored: %r'), key, exc)
-            config.intersphinx_mapping.pop(key)
-
-
 def setup(app: Sphinx) -> ExtensionMetadata:
     app.add_config_value('intersphinx_mapping', {}, 'env')
     app.add_config_value('intersphinx_cache_limit', 5, '')

sphinx/ext/intersphinx/_cli.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 
 import sys
 
-from sphinx.ext.intersphinx import fetch_inventory
+from sphinx.ext.intersphinx._load import fetch_inventory
 
 
 def inspect_main(argv: list[str], /) -> int:

sphinx/ext/intersphinx/_load.py

@@ -0,0 +1,252 @@
+"""This module contains the code for loading intersphinx inventories."""
+
+from __future__ import annotations
+
+import concurrent.futures
+import functools
+import posixpath
+import time
+from os import path
+from typing import TYPE_CHECKING
+from urllib.parse import urlsplit, urlunsplit
+
+from sphinx.builders.html import INVENTORY_FILENAME
+from sphinx.ext.intersphinx._shared import LOGGER as logger
+from sphinx.ext.intersphinx._shared import InventoryAdapter
+from sphinx.locale import __
+from sphinx.util import requests
+from sphinx.util.inventory import InventoryFile
+
+if TYPE_CHECKING:
+    from typing import IO
+
+    from sphinx.application import Sphinx
+    from sphinx.config import Config
+    from sphinx.ext.intersphinx._shared import InventoryCacheEntry
+    from sphinx.util.typing import Inventory
+
+
+def _strip_basic_auth(url: str) -> str:
+    """Returns *url* with basic auth credentials removed. Also returns the
+    basic auth username and password if they're present in *url*.
+
+    E.g.: https://user:pass@example.com => https://example.com
+
+    *url* need not include basic auth credentials.
+
+    :param url: url which may or may not contain basic auth credentials
+    :type url: ``str``
+
+    :return: *url* with any basic auth creds removed
+    :rtype: ``str``
+    """
+    frags = list(urlsplit(url))
+    # swap out "user[:pass]@hostname" for "hostname"
+    if '@' in frags[1]:
+        frags[1] = frags[1].split('@')[1]
+    return urlunsplit(frags)
+
+
+def _read_from_url(url: str, *, config: Config) -> IO:
+    """Reads data from *url* with an HTTP *GET*.
+
+    This function supports fetching from resources which use basic HTTP auth as
+    laid out by RFC1738 § 3.1. See § 5 for grammar definitions for URLs.
+
+    .. seealso:
+
+       https://www.ietf.org/rfc/rfc1738.txt
+
+    :param url: URL of an HTTP resource
+    :type url: ``str``
+
+    :return: data read from resource described by *url*
+    :rtype: ``file``-like object
+    """
+    r = requests.get(url, stream=True, timeout=config.intersphinx_timeout,
+                     _user_agent=config.user_agent,
+                     _tls_info=(config.tls_verify, config.tls_cacerts))
+    r.raise_for_status()
+    r.raw.url = r.url
+    # decode content-body based on the header.
+    # ref: https://github.com/psf/requests/issues/2155
+    r.raw.read = functools.partial(r.raw.read, decode_content=True)
+    return r.raw
+
+
+def _get_safe_url(url: str) -> str:
+    """Gets version of *url* with basic auth passwords obscured. This function
+    returns results suitable for printing and logging.
+
+    E.g.: https://user:12345@example.com => https://user@example.com
+
+    :param url: a url
+    :type url: ``str``
+
+    :return: *url* with password removed
+    :rtype: ``str``
+    """
+    parts = urlsplit(url)
+    if parts.username is None:
+        return url
+    else:
+        frags = list(parts)
+        if parts.port:
+            frags[1] = f'{parts.username}@{parts.hostname}:{parts.port}'
+        else:
+            frags[1] = f'{parts.username}@{parts.hostname}'
+
+        return urlunsplit(frags)
+
+
+def fetch_inventory(app: Sphinx, uri: str, inv: str) -> Inventory:
+    """Fetch, parse and return an intersphinx inventory file."""
+    # both *uri* (base URI of the links to generate) and *inv* (actual
+    # location of the inventory file) can be local or remote URIs
+    if '://' in uri:
+        # case: inv URI points to remote resource; strip any existing auth
+        uri = _strip_basic_auth(uri)
+    try:
+        if '://' in inv:
+            f = _read_from_url(inv, config=app.config)
+        else:
+            f = open(path.join(app.srcdir, inv), 'rb')  # NoQA: SIM115
+    except Exception as err:
+        err.args = ('intersphinx inventory %r not fetchable due to %s: %s',
+                    inv, err.__class__, str(err))
+        raise
+    try:
+        if hasattr(f, 'url'):
+            newinv = f.url
+            if inv != newinv:
+                logger.info(__('intersphinx inventory has moved: %s -> %s'), inv, newinv)
+
+                if uri in (inv, path.dirname(inv), path.dirname(inv) + '/'):
+                    uri = path.dirname(newinv)
+        with f:
+            try:
+                invdata = InventoryFile.load(f, uri, posixpath.join)
+            except ValueError as exc:
+                raise ValueError('unknown or unsupported inventory version: %r' % exc) from exc
+    except Exception as err:
+        err.args = ('intersphinx inventory %r not readable due to %s: %s',
+                    inv, err.__class__.__name__, str(err))
+        raise
+    else:
+        return invdata
+
+
+def fetch_inventory_group(
+    name: str | None,
+    uri: str,
+    invs: tuple[str | None, ...],
+    cache: dict[str, InventoryCacheEntry],
+    app: Sphinx,
+    now: int,
+) -> bool:
+    cache_time = now - app.config.intersphinx_cache_limit * 86400
+    failures = []
+    try:
+        for inv in invs:
+            if not inv:
+                inv = posixpath.join(uri, INVENTORY_FILENAME)
+            # decide whether the inventory must be read: always read local
+            # files; remote ones only if the cache time is expired
+            if '://' not in inv or uri not in cache or cache[uri][1] < cache_time:
+                safe_inv_url = _get_safe_url(inv)
+                logger.info(__('loading intersphinx inventory from %s...'), safe_inv_url)
+                try:
+                    invdata = fetch_inventory(app, uri, inv)
+                except Exception as err:
+                    failures.append(err.args)
+                    continue
+                if invdata:
+                    cache[uri] = name, now, invdata
+                    return True
+        return False
+    finally:
+        if failures == []:
+            pass
+        elif len(failures) < len(invs):
+            logger.info(__("encountered some issues with some of the inventories,"
+                           " but they had working alternatives:"))
+            for fail in failures:
+                logger.info(*fail)
+        else:
+            issues = '\n'.join(f[0] % f[1:] for f in failures)
+            logger.warning(__("failed to reach any of the inventories "
+                              "with the following issues:") + "\n" + issues)
+
+
+def load_mappings(app: Sphinx) -> None:
+    """Load all intersphinx mappings into the environment."""
+    now = int(time.time())
+    inventories = InventoryAdapter(app.builder.env)
+    intersphinx_cache: dict[str, InventoryCacheEntry] = inventories.cache
+
+    with concurrent.futures.ThreadPoolExecutor() as pool:
+        futures = []
+        name: str | None
+        uri: str
+        invs: tuple[str | None, ...]
+        for name, (uri, invs) in app.config.intersphinx_mapping.values():
+            futures.append(pool.submit(
+                fetch_inventory_group, name, uri, invs, intersphinx_cache, app, now,
+            ))
+        updated = [f.result() for f in concurrent.futures.as_completed(futures)]
+
+    if any(updated):
+        inventories.clear()
+
+        # Duplicate values in different inventories will shadow each
+        # other; which one will override which can vary between builds
+        # since they are specified using an unordered dict.  To make
+        # it more consistent, we sort the named inventories and then
+        # add the unnamed inventories last.  This means that the
+        # unnamed inventories will shadow the named ones but the named
+        # ones can still be accessed when the name is specified.
+        named_vals = []
+        unnamed_vals = []
+        for name, _expiry, invdata in intersphinx_cache.values():
+            if name:
+                named_vals.append((name, invdata))
+            else:
+                unnamed_vals.append((name, invdata))
+        for name, invdata in sorted(named_vals) + unnamed_vals:
+            if name:
+                inventories.named_inventory[name] = invdata
+            for type, objects in invdata.items():
+                inventories.main_inventory.setdefault(type, {}).update(objects)
+
+
+def normalize_intersphinx_mapping(app: Sphinx, config: Config) -> None:
+    for key, value in config.intersphinx_mapping.copy().items():
+        try:
+            if isinstance(value, (list, tuple)):
+                # new format
+                name, (uri, inv) = key, value
+                if not isinstance(name, str):
+                    logger.warning(__('intersphinx identifier %r is not string. Ignored'),
+                                   name)
+                    config.intersphinx_mapping.pop(key)
+                    continue
+            else:
+                # old format, no name
+                # xref RemovedInSphinx80Warning
+                name, uri, inv = None, key, value
+                msg = (
+                    "The pre-Sphinx 1.0 'intersphinx_mapping' format is "
+                    "deprecated and will be removed in Sphinx 8. Update to the "
+                    "current format as described in the documentation. "
+                    f"Hint: \"intersphinx_mapping = {{'<name>': {(uri, inv)!r}}}\"."
+                    "https://www.sphinx-doc.org/en/master/usage/extensions/intersphinx.html#confval-intersphinx_mapping"  # NoQA: E501
+                )
+                logger.warning(msg)
+
+            if not isinstance(inv, tuple):
+                config.intersphinx_mapping[key] = (name, (uri, (inv,)))
+            else:
+                config.intersphinx_mapping[key] = (name, (uri, inv))
+        except Exception as exc:
+            logger.warning(__('Failed to read intersphinx_mapping[%s], ignored: %r'), key, exc)
+            config.intersphinx_mapping.pop(key)

tests/test_extensions/test_ext_intersphinx.py

@@ -9,8 +9,6 @@ from docutils import nodes
 from sphinx import addnodes
 from sphinx.builders.html import INVENTORY_FILENAME
 from sphinx.ext.intersphinx import (
-    _get_safe_url,
-    _strip_basic_auth,
     fetch_inventory,
     inspect_main,
     load_mappings,
@@ -18,6 +16,7 @@ from sphinx.ext.intersphinx import (
     normalize_intersphinx_mapping,
 )
 from sphinx.ext.intersphinx import setup as intersphinx_setup
+from sphinx.ext.intersphinx._load import _get_safe_url, _strip_basic_auth
 from sphinx.util.console import strip_colors
 
 from tests.test_util.intersphinx_data import INVENTORY_V2, INVENTORY_V2_NO_VERSION
@@ -46,8 +45,8 @@ def set_config(app, mapping):
     app.config.intersphinx_disabled_reftypes = []
 
 
-@mock.patch('sphinx.ext.intersphinx.InventoryFile')
-@mock.patch('sphinx.ext.intersphinx._read_from_url')
+@mock.patch('sphinx.ext.intersphinx._load.InventoryFile')
+@mock.patch('sphinx.ext.intersphinx._load._read_from_url')
 def test_fetch_inventory_redirection(_read_from_url, InventoryFile, app, status, warning):  # NoQA: PT019
     intersphinx_setup(app)
     _read_from_url().readline.return_value = b'# Sphinx inventory version 2'