JezK
Edit File: xmlrpc.py
"""xmlrpclib.Transport implementation """ import logging import urllib.parse import xmlrpc.client from typing import TYPE_CHECKING, Tuple from pip._internal.exceptions import NetworkConnectionError from pip._internal.network.session import PipSession from pip._internal.network.utils import raise_for_status if TYPE_CHECKING: from xmlrpc.client import _HostType, _Marshallable logger = logging.getLogger(__name__) class PipXmlrpcTransport(xmlrpc.client.Transport): """Provide a `xmlrpclib.Transport` implementation via a `PipSession` object. """ def __init__( self, index_url: str, session: PipSession, use_datetime: bool = False ) -> None: super().__init__(use_datetime) index_parts = urllib.parse.urlparse(index_url) self._scheme = index_parts.scheme self._session = session def request( self, host: "_HostType", handler: str, request_body: bytes, verbose: bool = False, ) -> Tuple["_Marshallable", ...]: assert isinstance(host, str) parts = (self._scheme, host, handler, None, None, None) url = urllib.parse.urlunparse(parts) try: headers = {"Content-Type": "text/xml"} response = self._session.post( url, data=request_body, headers=headers, stream=True, ) raise_for_status(response) self.verbose = verbose return self.parse_response(response.raw) except NetworkConnectionError as exc: assert exc.response logger.critical( "HTTP error %s while getting %s", exc.response.status_code, url, ) raise