IntenseWebs/sphinx
3
0
Fork 0
mirror of https://github.com/sphinx-doc/sphinx.git synced 2025-02-25 18:55:22 -06:00
Code Issues Packages Projects Releases Wiki Activity

intersphinx: Add a type for project data (#12657)

Browse Source
This commit is contained in:
Adam Turner 2024-07-23 23:17:58 +01:00 committed by GitHub
parent 34bc4e69bb
commit 46671f2e10
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 179 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
sphinx/ext/intersphinx/__init__.py
View File

@ -21,7 +21,6 @@ from __future__ import annotations
__all__ = (
'InventoryAdapter',
'fetch_inventory',
'fetch_inventory_group',
'load_mappings',
'validate_intersphinx_mapping',
'IntersphinxRoleResolver',
@ -42,7 +41,6 @@ import sphinx
from sphinx.ext.intersphinx._cli import inspect_main
from sphinx.ext.intersphinx._load import (
fetch_inventory,
fetch_inventory_group,
load_mappings,
validate_intersphinx_mapping,
)

13
sphinx/ext/intersphinx/_cli.py
View File

@ -4,7 +4,7 @@ from __future__ import annotations
import sys
from sphinx.ext.intersphinx._load import fetch_inventory
from sphinx.ext.intersphinx._load import _fetch_inventory
def inspect_main(argv: list[str], /) -> int:
@ -21,13 +21,14 @@ def inspect_main(argv: list[str], /) -> int:
tls_cacerts: str | dict[str, str] | None = None
user_agent: str = ''
class MockApp:
srcdir = ''
config = MockConfig()
try:
filename = argv[0]
inv_data = fetch_inventory(MockApp(), '', filename) # type: ignore[arg-type]
inv_data = _fetch_inventory(
target_uri='',
inv_location=filename,
config=MockConfig(), # type: ignore[arg-type]
srcdir='' # type: ignore[arg-type]
)
for key in sorted(inv_data or {}):
print(key)
inv_entries = sorted(inv_data[key].items())

108
sphinx/ext/intersphinx/_load.py
View File

@ -13,12 +13,13 @@ from urllib.parse import urlsplit, urlunsplit
from sphinx.builders.html import INVENTORY_FILENAME
from sphinx.errors import ConfigError
from sphinx.ext.intersphinx._shared import LOGGER, InventoryAdapter
from sphinx.ext.intersphinx._shared import LOGGER, InventoryAdapter, _IntersphinxProject
from sphinx.locale import __
from sphinx.util import requests
from sphinx.util.inventory import InventoryFile
if TYPE_CHECKING:
from pathlib import Path
from typing import IO
from sphinx.application import Sphinx
@ -139,8 +140,17 @@ def load_mappings(app: Sphinx) -> None:
intersphinx_cache: dict[InventoryURI, InventoryCacheEntry] = inventories.cache
intersphinx_mapping: IntersphinxMapping = app.config.intersphinx_mapping
expected_uris = {uri for _name, (uri, _invs) in intersphinx_mapping.values()}
projects = []
for name, (uri, locations) in intersphinx_mapping.values():
try:
project = _IntersphinxProject(name=name, target_uri=uri, locations=locations)
except ValueError as err:
msg = __('An invalid intersphinx_mapping entry was added after normalisation.')
raise ConfigError(msg) from err
else:
projects.append(project)
expected_uris = {project.target_uri for project in projects}
for uri in frozenset(intersphinx_cache):
if intersphinx_cache[uri][0] not in intersphinx_mapping:
# Remove all cached entries that are no longer in `intersphinx_mapping`.
@ -153,8 +163,15 @@ def load_mappings(app: Sphinx) -> None:
with concurrent.futures.ThreadPoolExecutor() as pool:
futures = [
pool.submit(fetch_inventory_group, name, uri, invs, intersphinx_cache, app, now)
for name, (uri, invs) in app.config.intersphinx_mapping.values()
pool.submit(
_fetch_inventory_group,
project=project,
cache=intersphinx_cache,
now=now,
config=app.config,
srcdir=app.srcdir,
)
for project in projects
]
updated = [f.result() for f in concurrent.futures.as_completed(futures)]
@ -176,43 +193,52 @@ def load_mappings(app: Sphinx) -> None:
inventories.main_inventory.setdefault(objtype, {}).update(objects)
def fetch_inventory_group(
name: InventoryName,
uri: InventoryURI,
invs: tuple[InventoryLocation, ...],
def _fetch_inventory_group(
*,
project: _IntersphinxProject,
cache: dict[InventoryURI, InventoryCacheEntry],
app: Sphinx,
now: int,
config: Config,
srcdir: Path,
) -> bool:
cache_time = now - app.config.intersphinx_cache_limit * 86400
cache_time = now - config.intersphinx_cache_limit * 86400
updated = False
failures = []
for location in invs:
for location in project.locations:
# location is either None or a non-empty string
inv = f'{uri}/{INVENTORY_FILENAME}' if location is None else location
inv = f'{project.target_uri}/{INVENTORY_FILENAME}' if location is None else location
# 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:
if (
'://' not in inv
or project.target_uri not in cache
or cache[project.target_uri][1] < cache_time
):
LOGGER.info(__("loading intersphinx inventory '%s' from %s ..."),
name, _get_safe_url(inv))
project.name, _get_safe_url(inv))
try:
invdata = fetch_inventory(app, uri, inv)
invdata = _fetch_inventory(
target_uri=project.target_uri,
inv_location=inv,
config=config,
srcdir=srcdir,
)
except Exception as err:
failures.append(err.args)
continue
if invdata:
cache[uri] = name, now, invdata
cache[project.target_uri] = project.name, now, invdata
updated = True
break
if not failures:
pass
elif len(failures) < len(invs):
elif len(failures) < len(project.locations):
LOGGER.info(__('encountered some issues with some of the inventories,'
' but they had working alternatives:'))
for fail in failures:
@ -226,36 +252,54 @@ def fetch_inventory_group(
def fetch_inventory(app: Sphinx, uri: InventoryURI, 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:
return _fetch_inventory(
target_uri=uri,
inv_location=inv,
config=app.config,
srcdir=app.srcdir,
)
def _fetch_inventory(
*, target_uri: InventoryURI, inv_location: str, config: Config, srcdir: Path,
) -> Inventory:
"""Fetch, parse and return an intersphinx inventory file."""
# both *target_uri* (base URI of the links to generate)
# and *inv_location* (actual location of the inventory file)
# can be local or remote URIs
if '://' in target_uri:
# case: inv URI points to remote resource; strip any existing auth
uri = _strip_basic_auth(uri)
target_uri = _strip_basic_auth(target_uri)
try:
if '://' in inv:
f = _read_from_url(inv, config=app.config)
if '://' in inv_location:
f = _read_from_url(inv_location, config=config)
else:
f = open(path.join(app.srcdir, inv), 'rb') # NoQA: SIM115
f = open(path.join(srcdir, inv_location), 'rb') # NoQA: SIM115
except Exception as err:
err.args = ('intersphinx inventory %r not fetchable due to %s: %s',
inv, err.__class__, str(err))
inv_location, 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)
new_inv_location = f.url
if inv_location != new_inv_location:
msg = __('intersphinx inventory has moved: %s -> %s')
LOGGER.info(msg, inv_location, new_inv_location)
if uri in (inv, path.dirname(inv), path.dirname(inv) + '/'):
uri = path.dirname(newinv)
if target_uri in {
inv_location,
path.dirname(inv_location),
path.dirname(inv_location) + '/'
}:
target_uri = path.dirname(new_inv_location)
with f:
try:
invdata = InventoryFile.load(f, uri, posixpath.join)
invdata = InventoryFile.load(f, target_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))
inv_location, err.__class__.__name__, str(err))
raise
else:
return invdata

73
sphinx/ext/intersphinx/_shared.py
View File

@ -2,11 +2,12 @@
from __future__ import annotations
from typing import TYPE_CHECKING, Final
from typing import TYPE_CHECKING, Any, Final, NoReturn
from sphinx.util import logging
if TYPE_CHECKING:
from collections.abc import Sequence
from typing import TypeAlias
from sphinx.environment import BuildEnvironment
@ -31,7 +32,7 @@ if TYPE_CHECKING:
#: Inventory cache entry. The integer field is the cache expiration time.
InventoryCacheEntry: TypeAlias = tuple[InventoryName, int, Inventory]
#: The type of :confval:`intersphinx_mapping` *after* normalization.
#: The type of :confval:`intersphinx_mapping` *after* normalisation.
IntersphinxMapping = dict[
InventoryName,
tuple[InventoryName, tuple[InventoryURI, tuple[InventoryLocation, ...]]],
@ -40,6 +41,74 @@ if TYPE_CHECKING:
LOGGER: Final[logging.SphinxLoggerAdapter] = logging.getLogger('sphinx.ext.intersphinx')
class _IntersphinxProject:
name: InventoryName
target_uri: InventoryURI
locations: tuple[InventoryLocation, ...]
__slots__ = {
'name': 'The inventory name. '
'It is unique and in bijection with an remote inventory URL.',
'target_uri': 'The inventory project URL to which links are resolved. '
'It is unique and in bijection with an inventory name.',
'locations': 'A tuple of local or remote targets containing '
'the inventory data to fetch. '
'None indicates the default inventory file name.',
}
def __init__(
self,
*,
name: InventoryName,
target_uri: InventoryURI,
locations: Sequence[InventoryLocation],
) -> None:
if not name or not isinstance(name, str):
msg = 'name must be a non-empty string'
raise ValueError(msg)
if not target_uri or not isinstance(target_uri, str):
msg = 'target_uri must be a non-empty string'
raise ValueError(msg)
if not locations or not isinstance(locations, tuple):
msg = 'locations must be a non-empty tuple'
raise ValueError(msg)
if any(
location is not None and (not location or not isinstance(location, str))
for location in locations
):
msg = 'locations must be a tuple of strings or None'
raise ValueError(msg)
object.__setattr__(self, 'name', name)
object.__setattr__(self, 'target_uri', target_uri)
object.__setattr__(self, 'locations', tuple(locations))
def __repr__(self) -> str:
return (f'{self.__class__.__name__}('
f'name={self.name!r}, '
f'target_uri={self.target_uri!r}, '
f'locations={self.locations!r})')
def __eq__(self, other: object) -> bool:
if not isinstance(other, _IntersphinxProject):
return NotImplemented
return (
self.name == other.name
and self.target_uri == other.target_uri
and self.locations == other.locations
)
def __hash__(self) -> int:
return hash((self.name, self.target_uri, self.locations))
def __setattr__(self, key: str, value: Any) -> NoReturn:
msg = f'{self.__class__.__name__} is immutable'
raise AttributeError(msg)
def __delattr__(self, key: str) -> NoReturn:
msg = f'{self.__class__.__name__} is immutable'
raise AttributeError(msg)
class InventoryAdapter:
"""Inventory adapter for environment"""

31
tests/test_extensions/test_ext_intersphinx.py
View File

@ -13,14 +13,13 @@ from sphinx import addnodes
from sphinx.builders.html import INVENTORY_FILENAME
from sphinx.errors import ConfigError
from sphinx.ext.intersphinx import (
fetch_inventory,
inspect_main,
load_mappings,
missing_reference,
validate_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.ext.intersphinx._load import _fetch_inventory, _get_safe_url, _strip_basic_auth
from sphinx.util.console import strip_colors
from tests.test_util.intersphinx_data import (
@ -70,7 +69,12 @@ def test_fetch_inventory_redirection(_read_from_url, InventoryFile, app): # NoQ
# same uri and inv, not redirected
_read_from_url().url = 'https://hostname/' + INVENTORY_FILENAME
fetch_inventory(app, 'https://hostname/', 'https://hostname/' + INVENTORY_FILENAME)
_fetch_inventory(
target_uri='https://hostname/',
inv_location='https://hostname/' + INVENTORY_FILENAME,
config=app.config,
srcdir=app.srcdir,
)
assert 'intersphinx inventory has moved' not in app.status.getvalue()
assert InventoryFile.load.call_args[0][1] == 'https://hostname/'
@ -79,7 +83,12 @@ def test_fetch_inventory_redirection(_read_from_url, InventoryFile, app): # NoQ
app.status.truncate(0)
_read_from_url().url = 'https://hostname/new/' + INVENTORY_FILENAME
fetch_inventory(app, 'https://hostname/', 'https://hostname/' + INVENTORY_FILENAME)
_fetch_inventory(
target_uri='https://hostname/',
inv_location='https://hostname/' + INVENTORY_FILENAME,
config=app.config,
srcdir=app.srcdir,
)
assert app.status.getvalue() == ('intersphinx inventory has moved: '
'https://hostname/%s -> https://hostname/new/%s\n' %
(INVENTORY_FILENAME, INVENTORY_FILENAME))
@ -90,7 +99,12 @@ def test_fetch_inventory_redirection(_read_from_url, InventoryFile, app): # NoQ
app.status.truncate(0)
_read_from_url().url = 'https://hostname/new/' + INVENTORY_FILENAME
fetch_inventory(app, 'https://hostname/', 'https://hostname/new/' + INVENTORY_FILENAME)
_fetch_inventory(
target_uri='https://hostname/',
inv_location='https://hostname/new/' + INVENTORY_FILENAME,
config=app.config,
srcdir=app.srcdir,
)
assert 'intersphinx inventory has moved' not in app.status.getvalue()
assert InventoryFile.load.call_args[0][1] == 'https://hostname/'
@ -99,7 +113,12 @@ def test_fetch_inventory_redirection(_read_from_url, InventoryFile, app): # NoQ
app.status.truncate(0)
_read_from_url().url = 'https://hostname/other/' + INVENTORY_FILENAME
fetch_inventory(app, 'https://hostname/', 'https://hostname/new/' + INVENTORY_FILENAME)
_fetch_inventory(
target_uri='https://hostname/',
inv_location='https://hostname/new/' + INVENTORY_FILENAME,
config=app.config,
srcdir=app.srcdir,
)
assert app.status.getvalue() == ('intersphinx inventory has moved: '
'https://hostname/new/%s -> https://hostname/other/%s\n' %
(INVENTORY_FILENAME, INVENTORY_FILENAME))