Intersphinx, delegate to domains

Author: jakobandersen

sphinx/domains/__init__.py

@@ -11,8 +11,8 @@
 
 import copy
 from abc import ABC, abstractmethod
-from typing import (TYPE_CHECKING, Any, Callable, Dict, Iterable, List, NamedTuple, Tuple,
-                    Type, Union, cast)
+from typing import (TYPE_CHECKING, Any, Callable, Dict, Iterable, List, NamedTuple,
+                    Optional, Tuple, Type, Union, cast)
 
 from docutils import nodes
 from docutils.nodes import Element, Node, TextElement, system_message
@@ -22,6 +22,7 @@
 from sphinx.errors import SphinxError
 from sphinx.locale import _
 from sphinx.roles import XRefRole
+from sphinx.util.inventory import InventoryItemSet
 from sphinx.util.typing import RoleFunction
 
 if TYPE_CHECKING:
@@ -200,6 +201,12 @@ class Domain:
     #: data version, bump this when the format of `self.data` changes
     data_version = 0
 
+    #: intersphinx inventory value for an empty inventory
+    #: must be copy.deepcopy-able
+    #: if overridden, then the intersphinx methods in the domain should
+    #: probably also be overridden
+    initial_intersphinx_inventory = {}  # type: Dict
+
     def __init__(self, env: "BuildEnvironment") -> None:
         self.env = env              # type: BuildEnvironment
         self._role_cache = {}       # type: Dict[str, Callable]
@@ -318,7 +325,7 @@ def process_field_xref(self, pnode: pending_xref) -> None:
 
     def resolve_xref(self, env: "BuildEnvironment", fromdocname: str, builder: "Builder",
                      typ: str, target: str, node: pending_xref, contnode: TextElement
-                     ) -> Element:
+                     ) -> Optional[Element]:
         """Resolve the pending_xref *node* with the given *typ* and *target*.
 
         This method should return a new node, to replace the xref node,
@@ -401,3 +408,120 @@ def get_enumerable_node_type(self, node: Node) -> str:
     def get_full_qualified_name(self, node: Element) -> str:
         """Return full qualified name for given node."""
         return None
+
+    def intersphinx_add_entries_v2(self, store: Any,
+                                   data: Dict[Tuple[str, str], InventoryItemSet]) -> None:
+        """Store the given *data* for later intersphinx reference resolution.
+
+        This method is called at most once with all data loaded from inventories in
+        v1 and v2 format.
+
+        The *data* is a dictionary indexed by (object name, object type)
+        and must be stored in whichever way makes sense for the domain in the given *store*.
+        This *store* was initially a copy of :attr:`initial_intersphinx_inventory`.
+
+        The stored data is later given in :math:`intersphinx_resolve_xref` and
+        :meth:`intersphinx_resolve_xref_any` for reference resolution.
+
+        .. versionadded:: 4.0
+        """
+        store = cast(Dict[Tuple[str, str], InventoryItemSet], store)
+        store.update(data)
+
+    def _intersphinx_resolve_xref_lookup(self, store: Dict[Tuple[str, str], InventoryItemSet],
+                                         target: str, objtypes: List[str]
+                                         ) -> InventoryItemSet:
+        for objtype in objtypes:
+            if (target, objtype) not in store:
+                continue
+            return store[(target, objtype)]
+        return None
+
+    def _intersphinx_resolve_xref_2(self, store: Dict[Tuple[str, str], InventoryItemSet],
+                                    target: str, node: pending_xref,
+                                    objtypes: List[str]) -> InventoryItemSet:
+        # we try the target either as is, or with full qualification based on the scope of node
+        res = self._intersphinx_resolve_xref_lookup(store, target, objtypes)
+        if res is not None:
+            return res
+        # try with qualification of the current scope instead
+        # (we may have modified the target due to the splitting, so assign to the node)
+        old_target = node['reftarget']
+        node['reftarget'] = target
+        full_qualified_name = self.get_full_qualified_name(node)
+        node['reftarget'] = old_target
+        if full_qualified_name:
+            return self._intersphinx_resolve_xref_lookup(store, full_qualified_name, objtypes)
+        else:
+            return None
+
+    def _intersphinx_resolve_xref_1(self, store: Dict[Tuple[str, str], InventoryItemSet],
+                                    target: str, node: pending_xref,
+                                    contnode: TextElement, objtypes: List[str]
+                                    ) -> Optional[Element]:
+        # we adjust the object types for backwards compatibility
+        if self.name == 'std' and 'cmdoption' in objtypes:
+            # until Sphinx-1.6, cmdoptions are stored as std:option
+            objtypes.append('option')
+        if self.name == 'py' and 'attribute' in objtypes:
+            # Since Sphinx-2.1, properties are stored as py:method
+            objtypes.append('method')
+
+        # ordinary direct lookup
+        res = self._intersphinx_resolve_xref_2(store, target, node, objtypes)
+        if res is not None:
+            return res.make_refnode(self.name, target, node, contnode)
+
+        # try splitting the target into 'inv_name:target'
+        if ':' not in target:
+            return None
+        # first part may be the foreign doc set name
+        inv_name, newtarget = target.split(':', 1)
+        rawRes = self._intersphinx_resolve_xref_2(store, newtarget, node, objtypes)
+        if rawRes is None:
+            return None
+        rawRes = rawRes.select_inventory(inv_name)
+        if rawRes is None:
+            return None
+        # set the new target in the node so intersphinx can use it
+        node['reftarget'] = newtarget
+        return rawRes.make_refnode(self.name, target, node, contnode)
+
+    def intersphinx_resolve_xref(self, env: "BuildEnvironment", store: Any,
+                                 typ: str, target: str, node: pending_xref,
+                                 contnode: TextElement) -> Optional[Element]:
+        """Resolve the pending_xref *node* with the given *typ* and *target* via intersphinx.
+
+        This method should perform lookup in the given *store* which was created
+        through previous calls to :meth:`intersphinx_add_object`,
+        but otherwise behave very similarly to :meth:`resolve_xref`.
+
+        If a candidate is found in the store, a :class:`InventoryItemSet` is then a associated.
+        This item set should be used to create a new node, to replace the xref node,
+        containing the *contnode* which is the markup content of the cross-reference.
+
+        If no resolution can be found, None can be returned; and subsequent event handlers
+        will be given a chance to resolve the reference.
+        The method can also raise :exc:`sphinx.environment.NoUri` to suppress
+        any subsequent resolution of this reference.
+
+        .. versionadded:: 4.0
+        """
+        objtypes = self.objtypes_for_role(typ)
+        if not objtypes:
+            return None
+        return self._intersphinx_resolve_xref_1(store, target, node, contnode, objtypes)
+
+    def intersphinx_resolve_any_xref(self, env: "BuildEnvironment", store: Any,
+                                     target: str, node: pending_xref,
+                                     contnode: TextElement) -> Optional[Element]:
+        """Resolve the pending_xref *node* with the given *typ* and *target* via intersphinx.
+
+        The reference comes from an "any" or similar role, which means that we
+        don't know the type.
+        Otherwise, the arguments are the same as for :meth:`intersphinx_resolve_xref`.
+
+        .. versionadded:: 4.0
+        """
+        objtypes = list(self.object_types)
+        return self._intersphinx_resolve_xref_1(store, target, node, contnode, objtypes)

sphinx/ext/intersphinx.py

@@ -24,28 +24,26 @@
 """
 
 import concurrent.futures
+import copy
 import functools
 import posixpath
 import sys
 import time
 from os import path
-from typing import IO, Any, Dict, List, Tuple
+from typing import IO, Any, Dict, List, Optional, Tuple
 from urllib.parse import urlsplit, urlunsplit
 
-from docutils import nodes
-from docutils.nodes import TextElement
-from docutils.utils import relative_path
+from docutils.nodes import Element, TextElement
 
 import sphinx
 from sphinx.addnodes import pending_xref
 from sphinx.application import Sphinx
 from sphinx.builders.html import INVENTORY_FILENAME
 from sphinx.config import Config
 from sphinx.environment import BuildEnvironment
-from sphinx.locale import _, __
+from sphinx.locale import __
 from sphinx.util import logging, requests
-from sphinx.util.inventory import InventoryFile
-from sphinx.util.nodes import find_pending_xref_condition
+from sphinx.util.inventory import InventoryFile, InventoryItemSet
 from sphinx.util.typing import Inventory
 
 logger = logging.getLogger(__name__)
@@ -60,7 +58,13 @@ def __init__(self, env: BuildEnvironment) -> None:
         if not hasattr(env, 'intersphinx_cache'):
             self.env.intersphinx_cache = {}  # type: ignore
             self.env.intersphinx_inventory = {}  # type: ignore
-            self.env.intersphinx_named_inventory = {}  # type: ignore
+            self.env.intersphinx_by_domain_inventory = {}  # type: ignore
+            self._clear_by_domain_inventory()
+
+    def _clear_by_domain_inventory(self) -> None:
+        for domain in self.env.domains.values():
+            inv = copy.deepcopy(domain.initial_intersphinx_inventory)
+            self.env.intersphinx_by_domain_inventory[domain.name] = inv  # type: ignore
 
     @property
     def cache(self) -> Dict[str, Tuple[str, int, Inventory]]:
@@ -71,12 +75,13 @@ def main_inventory(self) -> Inventory:
         return self.env.intersphinx_inventory  # type: ignore
 
     @property
-    def named_inventory(self) -> Dict[str, Inventory]:
-        return self.env.intersphinx_named_inventory  # type: ignore
+    def by_domain_inventory(self) -> Dict[str, Any]:
+        return self.env.intersphinx_by_domain_inventory  # type: ignore
 
     def clear(self) -> None:
         self.env.intersphinx_inventory.clear()  # type: ignore
-        self.env.intersphinx_named_inventory.clear()  # type: ignore
+        self.env.intersphinx_by_domain_inventory.clear()  # type: ignore
+        self._clear_by_domain_inventory()
 
 
 def _strip_basic_auth(url: str) -> str:
@@ -241,114 +246,60 @@ def load_mappings(app: Sphinx) -> None:
 
     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.
+        # old stuff, still used in the tests
         cached_vals = list(inventories.cache.values())
         named_vals = sorted(v for v in cached_vals if v[0])
         unnamed_vals = [v for v in cached_vals if not v[0]]
         for name, _x, invdata in named_vals + unnamed_vals:
-            if name:
-                inventories.named_inventory[name] = invdata
             for type, objects in invdata.items():
                 inventories.main_inventory.setdefault(type, {}).update(objects)
+        # end of old stuff
+
+        # first collect all entries indexed by domain, object name, and object type
+        # domain -> (object_name, object_type) -> InventoryItemSet([(inv_name, inner_data)])
+        entries = {}  # type: Dict[str, Dict[Tuple[str, str], InventoryItemSet]]
+        for inv_name, _x, inv_data in inventories.cache.values():
+            # inv_data: Inventory = Dict[str, Dict[str, InventoryInner]]
+            for inv_type, inv_objects in inv_data.items():
+                # inv_objects: Dict[str, InventoryInner]
+                assert ':' in inv_type
+                domain_name, object_type = inv_type.split(':')
+                if domain_name not in app.env.domains:
+                    continue
+                domain_entries = entries.setdefault(domain_name, {})
+                for object_name, inner_data in inv_objects.items():
+                    itemSet = domain_entries.setdefault(
+                        (object_name, object_type), InventoryItemSet())
+                    itemSet.append((inv_name, inner_data))
+        # and then give the data to each domain
+        for domain_name, domain_entries in entries.items():
+            domain = app.env.domains[domain_name]
+            domain_store = inventories.by_domain_inventory[domain_name]
+            domain.intersphinx_add_entries_v2(domain_store, domain_entries)
 
 
 def missing_reference(app: Sphinx, env: BuildEnvironment, node: pending_xref,
-                      contnode: TextElement) -> nodes.reference:
+                      contnode: TextElement) -> Optional[Element]:
     """Attempt to resolve a missing reference via intersphinx references."""
+    typ = node['reftype']
     target = node['reftarget']
-    inventories = InventoryAdapter(env)
-    objtypes = None  # type: List[str]
-    if node['reftype'] == 'any':
-        # we search anything!
-        objtypes = ['%s:%s' % (domain.name, objtype)
-                    for domain in env.domains.values()
-                    for objtype in domain.object_types]
-        domain = None
+    inventories = InventoryAdapter(app.builder.env)
+    if typ == 'any':
+        for domain in env.domains.values():
+            domain_store = inventories.by_domain_inventory[domain.name]
+            res = domain.intersphinx_resolve_any_xref(
+                env, domain_store, target, node, contnode)
+            if res is not None:
+                return res
+        return None
     else:
-        domain = node.get('refdomain')
-        if not domain:
+        domain_name = node.get('refdomain')
+        if not domain_name:
             # only objects in domains are in the inventory
             return None
-        objtypes = env.get_domain(domain).objtypes_for_role(node['reftype'])
-        if not objtypes:
-            return None
-        objtypes = ['%s:%s' % (domain, objtype) for objtype in objtypes]
-    if 'std:cmdoption' in objtypes:
-        # until Sphinx-1.6, cmdoptions are stored as std:option
-        objtypes.append('std:option')
-    if 'py:attribute' in objtypes:
-        # Since Sphinx-2.1, properties are stored as py:method
-        objtypes.append('py:method')
-
-    # determine the contnode by pending_xref_condition
-    content = find_pending_xref_condition(node, 'resolved')
-    if content:
-        # resolved condition found.
-        contnodes = content.children
-        contnode = content.children[0]  # type: ignore
-    else:
-        # not resolved. Use the given contnode
-        contnodes = [contnode]
-
-    to_try = [(inventories.main_inventory, target)]
-    if domain:
-        full_qualified_name = env.get_domain(domain).get_full_qualified_name(node)
-        if full_qualified_name:
-            to_try.append((inventories.main_inventory, full_qualified_name))
-    in_set = None
-    if ':' in target:
-        # first part may be the foreign doc set name
-        setname, newtarget = target.split(':', 1)
-        if setname in inventories.named_inventory:
-            in_set = setname
-            to_try.append((inventories.named_inventory[setname], newtarget))
-            if domain:
-                node['reftarget'] = newtarget
-                full_qualified_name = env.get_domain(domain).get_full_qualified_name(node)
-                if full_qualified_name:
-                    to_try.append((inventories.named_inventory[setname], full_qualified_name))
-    for inventory, target in to_try:
-        for objtype in objtypes:
-            if objtype not in inventory or target not in inventory[objtype]:
-                continue
-            proj, version, uri, dispname = inventory[objtype][target]
-            if '://' not in uri and node.get('refdoc'):
-                # get correct path in case of subdirectories
-                uri = path.join(relative_path(node['refdoc'], '.'), uri)
-            if version:
-                reftitle = _('(in %s v%s)') % (proj, version)
-            else:
-                reftitle = _('(in %s)') % (proj,)
-            newnode = nodes.reference('', '', internal=False, refuri=uri, reftitle=reftitle)
-            if node.get('refexplicit'):
-                # use whatever title was given
-                newnode.extend(contnodes)
-            elif dispname == '-' or \
-                    (domain == 'std' and node['reftype'] == 'keyword'):
-                # use whatever title was given, but strip prefix
-                title = contnode.astext()
-                if in_set and title.startswith(in_set + ':'):
-                    newnode.append(contnode.__class__(title[len(in_set) + 1:],
-                                                      title[len(in_set) + 1:]))
-                else:
-                    newnode.extend(contnodes)
-            else:
-                # else use the given display name (used for :ref:)
-                newnode.append(contnode.__class__(dispname, dispname))
-            return newnode
-    # at least get rid of the ':' in the target if no explicit title given
-    if in_set is not None and not node.get('refexplicit', True):
-        if len(contnode) and isinstance(contnode[0], nodes.Text):
-            contnode[0] = nodes.Text(newtarget, contnode[0].rawsource)
-
-    return None
+        domain_store = inventories.by_domain_inventory[domain_name]
+        return env.domains[domain_name].intersphinx_resolve_xref(
+            env, domain_store, typ, target, node, contnode)
 
 
 def normalize_intersphinx_mapping(app: Sphinx, config: Config) -> None: