| """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 |
