pip._internal.index.collector Namespace Reference

Data Structures
class	_NotAPIContent
class	_NotHTTP
class	CacheablePageContent
class	CollectedSources
class	HTMLLinkParser
class	IndexContent
class	LinkCollector
class	ParseLinks

Functions
Optional[str]	_match_vcs_scheme (str url)
None	_ensure_api_header (Response response)
None	_ensure_api_response (str url, PipSession session)
Response	_get_simple_response (str url, PipSession session)
Optional[str]	_get_encoding_from_headers (ResponseHeaders headers)
ParseLinks	with_cached_index_content (ParseLinks fn)
Iterable[Link]	parse_links ("IndexContent" page)
None	_handle_get_simple_fail (Link link, Union[str, Exception] reason, Optional[Callable[..., None]] meth=None)
IndexContent	_make_index_content (Response response, bool cache_link_parsing=True)
Optional["IndexContent"]	_get_index_content (Link link, *PipSession session)

Variables
	Protocol = object
	logger = logging.getLogger(__name__)
	ResponseHeaders = MutableMapping[str, str]

Detailed Description

The main purpose of this module is to expose LinkCollector.collect_sources().

Function Documentation

_ensure_api_header()

None _ensure_api_header ( Response  response )

protected

Check the Content-Type header to ensure the response contains a Simple
API Response.

Raises `_NotAPIContent` if the content type is not a valid content-type.

Definition at line 73 of file collector.py.

def _ensure_api_header(response: Response) -> None:
    """
    Check the Content-Type header to ensure the response contains a Simple
    API Response.
 
    Raises `_NotAPIContent` if the content type is not a valid content-type.
    """
    content_type = response.headers.get("Content-Type", "Unknown")
 
    content_type_l = content_type.lower()
    if content_type_l.startswith(
        (
            "text/html",
            "application/vnd.pypi.simple.v1+html",
            "application/vnd.pypi.simple.v1+json",
        )
    ):
        return
 
    raise _NotAPIContent(content_type, response.request.method)
 
 

References i.

Referenced by pip._internal.index.collector._get_simple_response().

Here is the caller graph for this function:

_ensure_api_response()

None _ensure_api_response ( str  url, PipSession  session  )

protected

Send a HEAD request to the URL, and ensure the response contains a simple
API Response.

Raises `_NotHTTP` if the URL is not available for a HEAD request, or
`_NotAPIContent` if the content type is not a valid content type.

Definition at line 99 of file collector.py.

def _ensure_api_response(url: str, session: PipSession) -> None:
    """
    Send a HEAD request to the URL, and ensure the response contains a simple
    API Response.
 
    Raises `_NotHTTP` if the URL is not available for a HEAD request, or
    `_NotAPIContent` if the content type is not a valid content type.
    """
    scheme, netloc, path, query, fragment = urllib.parse.urlsplit(url)
    if scheme not in {"http", "https"}:
        raise _NotHTTP()
 
    resp = session.head(url, allow_redirects=True)
    raise_for_status(resp)
 
    _ensure_api_header(resp)
 
 

Referenced by pip._internal.index.collector._get_simple_response().

Here is the caller graph for this function:

_get_encoding_from_headers()

Optional[str] _get_encoding_from_headers ( ResponseHeaders  headers )

protected

Determine if we have any encoding information in our headers.

Definition at line 180 of file collector.py.

def _get_encoding_from_headers(headers: ResponseHeaders) -> Optional[str]:
    """Determine if we have any encoding information in our headers."""
    if headers and "Content-Type" in headers:
        m = email.message.Message()
        m["content-type"] = headers["Content-Type"]
        charset = m.get_param("charset")
        if charset:
            return str(charset)
    return None
 
 

References i.

Referenced by pip._internal.index.collector._make_index_content().

Here is the caller graph for this function:

_get_index_content()

Optional["IndexContent"] _get_index_content ( Link  link, *PipSession  session  )

protected

Definition at line 336 of file collector.py.

def _get_index_content(link: Link, *, session: PipSession) -> Optional["IndexContent"]:
    url = link.url.split("#", 1)[0]
 
    # Check for VCS schemes that do not support lookup as web pages.
    vcs_scheme = _match_vcs_scheme(url)
    if vcs_scheme:
        logger.warning(
            "Cannot look at %s URL %s because it does not support lookup as web pages.",
            vcs_scheme,
            link,
        )
        return None
 
    # Tack index.html onto file:// URLs that point to directories
    scheme, _, path, _, _, _ = urllib.parse.urlparse(url)
    if scheme == "file" and os.path.isdir(urllib.request.url2pathname(path)):
        # add trailing slash if not present so urljoin doesn't trim
        # final segment
        if not url.endswith("/"):
            url += "/"
        # TODO: In the future, it would be nice if pip supported PEP 691
        #       style responses in the file:// URLs, however there's no
        #       standard file extension for application/vnd.pypi.simple.v1+json
        #       so we'll need to come up with something on our own.
        url = urllib.parse.urljoin(url, "index.html")
        logger.debug(" file: URL is directory, getting %s", url)
 
    try:
        resp = _get_simple_response(url, session=session)
    except _NotHTTP:
        logger.warning(
            "Skipping page %s because it looks like an archive, and cannot "
            "be checked by a HTTP HEAD request.",
            link,
        )
    except _NotAPIContent as exc:
        logger.warning(
            "Skipping page %s because the %s request got Content-Type: %s. "
            "The only supported Content-Types are application/vnd.pypi.simple.v1+json, "
            "application/vnd.pypi.simple.v1+html, and text/html",
            link,
            exc.request_desc,
            exc.content_type,
        )
    except NetworkConnectionError as exc:
        _handle_get_simple_fail(link, exc)
    except RetryError as exc:
        _handle_get_simple_fail(link, exc)
    except SSLError as exc:
        reason = "There was a problem confirming the ssl certificate: "
        reason += str(exc)
        _handle_get_simple_fail(link, reason, meth=logger.info)
    except requests.ConnectionError as exc:
        _handle_get_simple_fail(link, f"connection error: {exc}")
    except requests.Timeout:
        _handle_get_simple_fail(link, "timed out")
    else:
        return _make_index_content(resp, cache_link_parsing=link.cache_link_parsing)
    return None
 
 

References pip._internal.index.collector._get_simple_response(), pip._internal.index.collector._handle_get_simple_fail(), pip._internal.index.collector._make_index_content(), pip._internal.index.collector._match_vcs_scheme(), and i.

Referenced by LinkCollector.fetch_response().

Here is the call graph for this function:

Here is the caller graph for this function:

_get_simple_response()

Response _get_simple_response ( str  url, PipSession  session  )

protected

Access an Simple API response with GET, and return the response.

This consists of three parts:

1. If the URL looks suspiciously like an archive, send a HEAD first to
   check the Content-Type is HTML or Simple API, to avoid downloading a
   large file. Raise `_NotHTTP` if the content type cannot be determined, or
   `_NotAPIContent` if it is not HTML or a Simple API.
2. Actually perform the request. Raise HTTP exceptions on network failures.
3. Check the Content-Type header to make sure we got a Simple API response,
   and raise `_NotAPIContent` otherwise.

Definition at line 117 of file collector.py.

def _get_simple_response(url: str, session: PipSession) -> Response:
    """Access an Simple API response with GET, and return the response.
 
    This consists of three parts:
 
    1. If the URL looks suspiciously like an archive, send a HEAD first to
       check the Content-Type is HTML or Simple API, to avoid downloading a
       large file. Raise `_NotHTTP` if the content type cannot be determined, or
       `_NotAPIContent` if it is not HTML or a Simple API.
    2. Actually perform the request. Raise HTTP exceptions on network failures.
    3. Check the Content-Type header to make sure we got a Simple API response,
       and raise `_NotAPIContent` otherwise.
    """
    if is_archive_file(Link(url).filename):
        _ensure_api_response(url, session=session)
 
    logger.debug("Getting page %s", redact_auth_from_url(url))
 
    resp = session.get(
        url,
        headers={
            "Accept": ", ".join(
                [
                    "application/vnd.pypi.simple.v1+json",
                    "application/vnd.pypi.simple.v1+html; q=0.1",
                    "text/html; q=0.01",
                ]
            ),
            # We don't want to blindly returned cached data for
            # /simple/, because authors generally expecting that
            # twine upload && pip install will function, but if
            # they've done a pip install in the last ~10 minutes
            # it won't. Thus by setting this to zero we will not
            # blindly use any cached data, however the benefit of
            # using max-age=0 instead of no-cache, is that we will
            # still support conditional requests, so we will still
            # minimize traffic sent in cases where the page hasn't
            # changed at all, we will just always incur the round
            # trip for the conditional GET now instead of only
            # once per 10 minutes.
            # For more information, please see pypa/pip#5670.
            "Cache-Control": "max-age=0",
        },
    )
    raise_for_status(resp)
 
    # The check for archives above only works if the url ends with
    # something that looks like an archive. However that is not a
    # requirement of an url. Unless we issue a HEAD request on every
    # url we cannot know ahead of time for sure if something is a
    # Simple API response or not. However we can check after we've
    # downloaded it.
    _ensure_api_header(resp)
 
    logger.debug(
        "Fetched page %s as %s",
        redact_auth_from_url(url),
        resp.headers.get("Content-Type", "Unknown"),
    )
 
    return resp
 
 

References pip._internal.index.collector._ensure_api_header(), pip._internal.index.collector._ensure_api_response(), and i.

Referenced by pip._internal.index.collector._get_index_content().

Here is the call graph for this function:

Here is the caller graph for this function:

_handle_get_simple_fail()

None _handle_get_simple_fail ( Link  link, Union[str, Exception]  reason, Optional[Callable[..., None]]   meth = None  )

protected

Definition at line 313 of file collector.py.

) -> None:
    if meth is None:
        meth = logger.debug
    meth("Could not fetch URL %s: %s - skipping", link, reason)
 
 

References i.

Referenced by pip._internal.index.collector._get_index_content().

Here is the caller graph for this function:

_make_index_content()

IndexContent _make_index_content ( Response  response, bool   cache_link_parsing = True  )

protected

Definition at line 323 of file collector.py.

) -> IndexContent:
    encoding = _get_encoding_from_headers(response.headers)
    return IndexContent(
        response.content,
        response.headers["Content-Type"],
        encoding=encoding,
        url=response.url,
        cache_link_parsing=cache_link_parsing,
    )
 
 

References pip._internal.index.collector._get_encoding_from_headers(), and i.

Referenced by pip._internal.index.collector._get_index_content().

Here is the call graph for this function:

Here is the caller graph for this function:

_match_vcs_scheme()

Optional[str] _match_vcs_scheme ( str  url )

protected

Look for VCS schemes in the URL.

Returns the matched VCS scheme, or None if there's no match.

Definition at line 55 of file collector.py.

def _match_vcs_scheme(url: str) -> Optional[str]:
    """Look for VCS schemes in the URL.
 
    Returns the matched VCS scheme, or None if there's no match.
    """
    for scheme in vcs.schemes:
        if url.lower().startswith(scheme) and url[len(scheme)] in "+:":
            return scheme
    return None
 
 

References i.

Referenced by pip._internal.index.collector._get_index_content().

Here is the caller graph for this function:

parse_links()

Iterable[Link] parse_links

(

"IndexContent"

page

)

Parse a Simple API's Index Content, and yield its anchor elements as Link objects.

Definition at line 229 of file collector.py.

def parse_links(page: "IndexContent") -> Iterable[Link]:
    """
    Parse a Simple API's Index Content, and yield its anchor elements as Link objects.
    """
 
    content_type_l = page.content_type.lower()
    if content_type_l.startswith("application/vnd.pypi.simple.v1+json"):
        data = json.loads(page.content)
        for file in data.get("files", []):
            link = Link.from_json(file, page.url)
            if link is None:
                continue
            yield link
        return
 
    parser = HTMLLinkParser(page.url)
    encoding = page.encoding or "utf-8"
    parser.feed(page.content.decode(encoding))
 
    url = page.url
    base_url = parser.base_url or url
    for anchor in parser.anchors:
        link = Link.from_element(anchor, page_url=url, base_url=base_url)
        if link is None:
            continue
        yield link
 
 

References i.

with_cached_index_content()

ParseLinks with_cached_index_content

(

ParseLinks

fn

)

Given a function that parses an Iterable[Link] from an IndexContent, cache the
function's result (keyed by CacheablePageContent), unless the IndexContent
`page` has `page.cache_link_parsing == False`.

Definition at line 208 of file collector.py.

def with_cached_index_content(fn: ParseLinks) -> ParseLinks:
    """
    Given a function that parses an Iterable[Link] from an IndexContent, cache the
    function's result (keyed by CacheablePageContent), unless the IndexContent
    `page` has `page.cache_link_parsing == False`.
    """
 
    @functools.lru_cache(maxsize=None)
    def wrapper(cacheable_page: CacheablePageContent) -> List[Link]:
        return list(fn(cacheable_page.page))
 
    @functools.wraps(fn)
    def wrapper_wrapper(page: "IndexContent") -> List[Link]:
        if page.cache_link_parsing:
            return wrapper(CacheablePageContent(page))
        return list(fn(page))
 
    return wrapper_wrapper
 
 
@with_cached_index_content

References i.

Variable Documentation

logger

logger = logging.getLogger(__name__)

Definition at line 50 of file collector.py.

Protocol

Protocol = object

Definition at line 48 of file collector.py.

ResponseHeaders

ResponseHeaders = MutableMapping[str, str]

Definition at line 52 of file collector.py.