We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Improve proxy connection failure error message
Raise a more informative error message when the proxy rejects requests from Pulp

BZ: https://bugzilla.redhat.com/show_bug.cgi?id=2047485

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `pulpcore/download/http.py`
Content:
```
1 import logging
2 
3 import aiohttp
4 import asyncio
5 import backoff
6 
7 from .base import BaseDownloader, DownloadResult
8 from pulpcore.exceptions import (
9     DigestValidationError,
10     SizeValidationError,
11     TimeoutException,
12 )
13 
14 
15 log = logging.getLogger(__name__)
16 
17 
18 logging.getLogger("backoff").addHandler(logging.StreamHandler())
19 
20 
21 def http_giveup_handler(exc):
22     """
23     Inspect a raised exception and determine if we should give up.
24 
25     Do not give up when the error is one of the following:
26 
27         HTTP 429 - Too Many Requests
28         HTTP 5xx - Server errors
29         Socket timeout
30         TCP disconnect
31         Client SSL Error
32 
33     Based on the AWS and Google Cloud guidelines:
34         https://docs.aws.amazon.com/general/latest/gr/api-retries.html
35         https://cloud.google.com/storage/docs/retry-strategy
36 
37     Args:
38         exc (Exception): The exception to inspect
39 
40     Returns:
41         True if the download should give up, False otherwise
42     """
43     if isinstance(exc, aiohttp.ClientResponseError):
44         server_error = 500 <= exc.code < 600
45         too_many_requests = exc.code == 429
46         return not server_error and not too_many_requests
47 
48     # any other type of error (pre-filtered by the backoff decorator) shouldn't be fatal
49     return False
50 
51 
52 class HttpDownloader(BaseDownloader):
53     """
54     An HTTP/HTTPS Downloader built on `aiohttp`.
55 
56     This downloader downloads data from one `url` and is not reused.
57 
58     The downloader optionally takes a session argument, which is an `aiohttp.ClientSession`. This
59     allows many downloaders to share one `aiohttp.ClientSession` which provides a connection pool,
60     connection reuse, and keep-alives across multiple downloaders. When creating many downloaders,
61     have one session shared by all of your `HttpDownloader` objects.
62 
63     A session is optional; if omitted, one session will be created, used for this downloader, and
64     then closed when the download is complete. A session that is passed in will not be closed when
65     the download is complete.
66 
67     If a session is not provided, the one created by HttpDownloader uses non-default timing values.
68     Specifically, the "total" timeout is set to None and the "sock_connect" and "sock_read" are both
69     5 minutes. For more info on these settings, see the aiohttp docs:
70     http://aiohttp.readthedocs.io/en/stable/client_quickstart.html#timeouts Behaviorally, it should
71     allow for an active download to be arbitrarily long, while still detecting dead or closed
72     sessions even when TCPKeepAlive is disabled.
73 
74     If a session is not provided, the one created will force TCP connection closure after each
75     request. This is done for compatibility reasons due to various issues related to session
76     continuation implementation in various servers.
77 
78     `aiohttp.ClientSession` objects allows you to configure options that will apply to all
79     downloaders using that session such as auth, timeouts, headers, etc. For more info on these
80     options see the `aiohttp.ClientSession` docs for more information:
81     http://aiohttp.readthedocs.io/en/stable/client_reference.html#aiohttp.ClientSession
82 
83     The `aiohttp.ClientSession` can additionally be configured for SSL configuration by passing in a
84     `aiohttp.TCPConnector`. For information on configuring either server or client certificate based
85     identity verification, see the aiohttp documentation:
86     http://aiohttp.readthedocs.io/en/stable/client.html#ssl-control-for-tcp-sockets
87 
88     For more information on `aiohttp.BasicAuth` objects, see their docs:
89     http://aiohttp.readthedocs.io/en/stable/client_reference.html#aiohttp.BasicAuth
90 
91     Synchronous Download::
92 
93         downloader = HttpDownloader('http://example.com/')
94         result = downloader.fetch()
95 
96     Parallel Download::
97 
98         download_coroutines = [
99             HttpDownloader('http://example.com/').run(),
100             HttpDownloader('http://pulpproject.org/').run(),
101         ]
102 
103         loop = asyncio.get_event_loop()
104         done, not_done = loop.run_until_complete(asyncio.wait(download_coroutines))
105 
106         for task in done:
107             try:
108                 task.result()  # This is a DownloadResult
109             except Exception as error:
110                 pass  # fatal exceptions are raised by result()
111 
112     The HTTPDownloaders contain automatic retry logic if the server responds with HTTP 429 response.
113     The coroutine will automatically retry 10 times with exponential backoff before allowing a
114     final exception to be raised.
115 
116     Attributes:
117         session (aiohttp.ClientSession): The session to be used by the downloader.
118         auth (aiohttp.BasicAuth): An object that represents HTTP Basic Authorization or None
119         proxy (str): An optional proxy URL or None
120         proxy_auth (aiohttp.BasicAuth): An optional object that represents proxy HTTP Basic
121             Authorization or None
122         headers_ready_callback (callable): An optional callback that accepts a single dictionary
123             as its argument. The callback will be called when the response headers are
124             available. The dictionary passed has the header names as the keys and header values
125             as its values. e.g. `{'Transfer-Encoding': 'chunked'}`. This can also be None.
126 
127     This downloader also has all of the attributes of
128     :class:`~pulpcore.plugin.download.BaseDownloader`
129     """
130 
131     def __init__(
132         self,
133         url,
134         session=None,
135         auth=None,
136         proxy=None,
137         proxy_auth=None,
138         headers_ready_callback=None,
139         headers=None,
140         throttler=None,
141         max_retries=0,
142         **kwargs,
143     ):
144         """
145         Args:
146             url (str): The url to download.
147             session (aiohttp.ClientSession): The session to be used by the downloader. (optional) If
148                 not specified it will open the session and close it
149             auth (aiohttp.BasicAuth): An object that represents HTTP Basic Authorization (optional)
150             proxy (str): An optional proxy URL.
151             proxy_auth (aiohttp.BasicAuth): An optional object that represents proxy HTTP Basic
152                 Authorization.
153             headers_ready_callback (callable): An optional callback that accepts a single dictionary
154                 as its argument. The callback will be called when the response headers are
155                 available. The dictionary passed has the header names as the keys and header values
156                 as its values. e.g. `{'Transfer-Encoding': 'chunked'}`
157             headers (dict): Headers to be submitted with the request.
158             throttler (asyncio_throttle.Throttler): Throttler for asyncio.
159             max_retries (int): The maximum number of times to retry a download upon failure.
160             kwargs (dict): This accepts the parameters of
161                 :class:`~pulpcore.plugin.download.BaseDownloader`.
162         """
163         if session:
164             self.session = session
165             self._close_session_on_finalize = False
166         else:
167             timeout = aiohttp.ClientTimeout(total=None, sock_connect=600, sock_read=600)
168             conn = aiohttp.TCPConnector({"force_close": True})
169             self.session = aiohttp.ClientSession(connector=conn, timeout=timeout, headers=headers)
170             self._close_session_on_finalize = True
171         self.auth = auth
172         self.proxy = proxy
173         self.proxy_auth = proxy_auth
174         self.headers_ready_callback = headers_ready_callback
175         self.download_throttler = throttler
176         self.max_retries = max_retries
177         super().__init__(url, **kwargs)
178 
179     def raise_for_status(self, response):
180         """
181         Raise error if aiohttp response status is >= 400 and not silenced.
182 
183         Args:
184             response (aiohttp.ClientResponse): The response to handle.
185 
186         Raises:
187                aiohttp.ClientResponseError: When the response status is >= 400.
188         """
189         response.raise_for_status()
190 
191     async def _handle_response(self, response):
192         """
193         Handle the aiohttp response by writing it to disk and calculating digests
194 
195         Args:
196             response (aiohttp.ClientResponse): The response to handle.
197 
198         Returns:
199              DownloadResult: Contains information about the result. See the DownloadResult docs for
200                  more information.
201         """
202         if self.headers_ready_callback:
203             await self.headers_ready_callback(response.headers)
204         while True:
205             chunk = await response.content.read(1048576)  # 1 megabyte
206             if not chunk:
207                 await self.finalize()
208                 break  # the download is done
209             await self.handle_data(chunk)
210         return DownloadResult(
211             path=self.path,
212             artifact_attributes=self.artifact_attributes,
213             url=self.url,
214             headers=response.headers,
215         )
216 
217     async def run(self, extra_data=None):
218         """
219         Run the downloader with concurrency restriction and retry logic.
220 
221         This method acquires `self.semaphore` before calling the actual download implementation
222         contained in `_run()`. This ensures that the semaphore stays acquired even as the `backoff`
223         wrapper around `_run()`, handles backoff-and-retry logic.
224 
225         Args:
226             extra_data (dict): Extra data passed to the downloader.
227 
228         Returns:
229             :class:`~pulpcore.plugin.download.DownloadResult` from `_run()`.
230 
231         """
232         retryable_errors = (
233             aiohttp.ClientConnectorSSLError,
234             aiohttp.ClientConnectorError,
235             aiohttp.ClientOSError,
236             aiohttp.ClientPayloadError,
237             aiohttp.ClientResponseError,
238             aiohttp.ServerDisconnectedError,
239             TimeoutError,
240             TimeoutException,
241             DigestValidationError,
242             SizeValidationError,
243         )
244 
245         async with self.semaphore:
246 
247             @backoff.on_exception(
248                 backoff.expo,
249                 retryable_errors,
250                 max_tries=self.max_retries + 1,
251                 giveup=http_giveup_handler,
252             )
253             async def download_wrapper():
254                 self._ensure_no_broken_file()
255                 try:
256                     return await self._run(extra_data=extra_data)
257                 except asyncio.TimeoutError:
258                     raise TimeoutException(self.url)
259 
260             return await download_wrapper()
261 
262     async def _run(self, extra_data=None):
263         """
264         Download, validate, and compute digests on the `url`. This is a coroutine.
265 
266         This method is externally wrapped with backoff-and-retry behavior for some errors.
267         It retries with exponential backoff some number of times before allowing a final
268         exception to be raised.
269 
270         This method provides the same return object type and documented in
271         :meth:`~pulpcore.plugin.download.BaseDownloader._run`.
272 
273         Args:
274             extra_data (dict): Extra data passed by the downloader.
275         """
276         if self.download_throttler:
277             await self.download_throttler.acquire()
278         async with self.session.get(
279             self.url, proxy=self.proxy, proxy_auth=self.proxy_auth, auth=self.auth
280         ) as response:
281             self.raise_for_status(response)
282             to_return = await self._handle_response(response)
283             await response.release()
284         if self._close_session_on_finalize:
285             await self.session.close()
286         return to_return
287 
288     def _ensure_no_broken_file(self):
289         """Upon retry reset writer back to None to get a fresh file."""
290         if self._writer is not None:
291             self._writer.delete = True
292             self._writer.close()
293             self._writer = None
294
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/pulpcore/download/http.py b/pulpcore/download/http.py
--- a/pulpcore/download/http.py
+++ b/pulpcore/download/http.py
@@ -256,6 +256,17 @@
                     return await self._run(extra_data=extra_data)
                 except asyncio.TimeoutError:
                     raise TimeoutException(self.url)
+                except aiohttp.ClientHttpProxyError as e:
+                    log.error(
+                        "Proxy {!r} rejected connection request during a request to "
+                        "{!r}, status={}, message={!r}".format(
+                            e.request_info.real_url,
+                            e.request_info.url,
+                            e.status,
+                            e.message,
+                        )
+                    )
+                    raise e
 
             return await download_wrapper()

{"golden_diff": "diff --git a/pulpcore/download/http.py b/pulpcore/download/http.py\n--- a/pulpcore/download/http.py\n+++ b/pulpcore/download/http.py\n@@ -256,6 +256,17 @@\n                     return await self._run(extra_data=extra_data)\n                 except asyncio.TimeoutError:\n                     raise TimeoutException(self.url)\n+                except aiohttp.ClientHttpProxyError as e:\n+                    log.error(\n+                        \"Proxy {!r} rejected connection request during a request to \"\n+                        \"{!r}, status={}, message={!r}\".format(\n+                            e.request_info.real_url,\n+                            e.request_info.url,\n+                            e.status,\n+                            e.message,\n+                        )\n+                    )\n+                    raise e\n \n             return await download_wrapper()\n", "issue": "Improve proxy connection failure error message\nRaise a more informative error message when the proxy rejects requests from Pulp\r\n\r\nBZ: https://bugzilla.redhat.com/show_bug.cgi?id=2047485\n", "before_files": [{"content": "import logging\n\nimport aiohttp\nimport asyncio\nimport backoff\n\nfrom .base import BaseDownloader, DownloadResult\nfrom pulpcore.exceptions import (\n    DigestValidationError,\n    SizeValidationError,\n    TimeoutException,\n)\n\n\nlog = logging.getLogger(__name__)\n\n\nlogging.getLogger(\"backoff\").addHandler(logging.StreamHandler())\n\n\ndef http_giveup_handler(exc):\n    \"\"\"\n    Inspect a raised exception and determine if we should give up.\n\n    Do not give up when the error is one of the following:\n\n        HTTP 429 - Too Many Requests\n        HTTP 5xx - Server errors\n        Socket timeout\n        TCP disconnect\n        Client SSL Error\n\n    Based on the AWS and Google Cloud guidelines:\n        https://docs.aws.amazon.com/general/latest/gr/api-retries.html\n        https://cloud.google.com/storage/docs/retry-strategy\n\n    Args:\n        exc (Exception): The exception to inspect\n\n    Returns:\n        True if the download should give up, False otherwise\n    \"\"\"\n    if isinstance(exc, aiohttp.ClientResponseError):\n        server_error = 500 <= exc.code < 600\n        too_many_requests = exc.code == 429\n        return not server_error and not too_many_requests\n\n    # any other type of error (pre-filtered by the backoff decorator) shouldn't be fatal\n    return False\n\n\nclass HttpDownloader(BaseDownloader):\n    \"\"\"\n    An HTTP/HTTPS Downloader built on `aiohttp`.\n\n    This downloader downloads data from one `url` and is not reused.\n\n    The downloader optionally takes a session argument, which is an `aiohttp.ClientSession`. This\n    allows many downloaders to share one `aiohttp.ClientSession` which provides a connection pool,\n    connection reuse, and keep-alives across multiple downloaders. When creating many downloaders,\n    have one session shared by all of your `HttpDownloader` objects.\n\n    A session is optional; if omitted, one session will be created, used for this downloader, and\n    then closed when the download is complete. A session that is passed in will not be closed when\n    the download is complete.\n\n    If a session is not provided, the one created by HttpDownloader uses non-default timing values.\n    Specifically, the \"total\" timeout is set to None and the \"sock_connect\" and \"sock_read\" are both\n    5 minutes. For more info on these settings, see the aiohttp docs:\n    http://aiohttp.readthedocs.io/en/stable/client_quickstart.html#timeouts Behaviorally, it should\n    allow for an active download to be arbitrarily long, while still detecting dead or closed\n    sessions even when TCPKeepAlive is disabled.\n\n    If a session is not provided, the one created will force TCP connection closure after each\n    request. This is done for compatibility reasons due to various issues related to session\n    continuation implementation in various servers.\n\n    `aiohttp.ClientSession` objects allows you to configure options that will apply to all\n    downloaders using that session such as auth, timeouts, headers, etc. For more info on these\n    options see the `aiohttp.ClientSession` docs for more information:\n    http://aiohttp.readthedocs.io/en/stable/client_reference.html#aiohttp.ClientSession\n\n    The `aiohttp.ClientSession` can additionally be configured for SSL configuration by passing in a\n    `aiohttp.TCPConnector`. For information on configuring either server or client certificate based\n    identity verification, see the aiohttp documentation:\n    http://aiohttp.readthedocs.io/en/stable/client.html#ssl-control-for-tcp-sockets\n\n    For more information on `aiohttp.BasicAuth` objects, see their docs:\n    http://aiohttp.readthedocs.io/en/stable/client_reference.html#aiohttp.BasicAuth\n\n    Synchronous Download::\n\n        downloader = HttpDownloader('http://example.com/')\n        result = downloader.fetch()\n\n    Parallel Download::\n\n        download_coroutines = [\n            HttpDownloader('http://example.com/').run(),\n            HttpDownloader('http://pulpproject.org/').run(),\n        ]\n\n        loop = asyncio.get_event_loop()\n        done, not_done = loop.run_until_complete(asyncio.wait(download_coroutines))\n\n        for task in done:\n            try:\n                task.result()  # This is a DownloadResult\n            except Exception as error:\n                pass  # fatal exceptions are raised by result()\n\n    The HTTPDownloaders contain automatic retry logic if the server responds with HTTP 429 response.\n    The coroutine will automatically retry 10 times with exponential backoff before allowing a\n    final exception to be raised.\n\n    Attributes:\n        session (aiohttp.ClientSession): The session to be used by the downloader.\n        auth (aiohttp.BasicAuth): An object that represents HTTP Basic Authorization or None\n        proxy (str): An optional proxy URL or None\n        proxy_auth (aiohttp.BasicAuth): An optional object that represents proxy HTTP Basic\n            Authorization or None\n        headers_ready_callback (callable): An optional callback that accepts a single dictionary\n            as its argument. The callback will be called when the response headers are\n            available. The dictionary passed has the header names as the keys and header values\n            as its values. e.g. `{'Transfer-Encoding': 'chunked'}`. This can also be None.\n\n    This downloader also has all of the attributes of\n    :class:`~pulpcore.plugin.download.BaseDownloader`\n    \"\"\"\n\n    def __init__(\n        self,\n        url,\n        session=None,\n        auth=None,\n        proxy=None,\n        proxy_auth=None,\n        headers_ready_callback=None,\n        headers=None,\n        throttler=None,\n        max_retries=0,\n        **kwargs,\n    ):\n        \"\"\"\n        Args:\n            url (str): The url to download.\n            session (aiohttp.ClientSession): The session to be used by the downloader. (optional) If\n                not specified it will open the session and close it\n            auth (aiohttp.BasicAuth): An object that represents HTTP Basic Authorization (optional)\n            proxy (str): An optional proxy URL.\n            proxy_auth (aiohttp.BasicAuth): An optional object that represents proxy HTTP Basic\n                Authorization.\n            headers_ready_callback (callable): An optional callback that accepts a single dictionary\n                as its argument. The callback will be called when the response headers are\n                available. The dictionary passed has the header names as the keys and header values\n                as its values. e.g. `{'Transfer-Encoding': 'chunked'}`\n            headers (dict): Headers to be submitted with the request.\n            throttler (asyncio_throttle.Throttler): Throttler for asyncio.\n            max_retries (int): The maximum number of times to retry a download upon failure.\n            kwargs (dict): This accepts the parameters of\n                :class:`~pulpcore.plugin.download.BaseDownloader`.\n        \"\"\"\n        if session:\n            self.session = session\n            self._close_session_on_finalize = False\n        else:\n            timeout = aiohttp.ClientTimeout(total=None, sock_connect=600, sock_read=600)\n            conn = aiohttp.TCPConnector({\"force_close\": True})\n            self.session = aiohttp.ClientSession(connector=conn, timeout=timeout, headers=headers)\n            self._close_session_on_finalize = True\n        self.auth = auth\n        self.proxy = proxy\n        self.proxy_auth = proxy_auth\n        self.headers_ready_callback = headers_ready_callback\n        self.download_throttler = throttler\n        self.max_retries = max_retries\n        super().__init__(url, **kwargs)\n\n    def raise_for_status(self, response):\n        \"\"\"\n        Raise error if aiohttp response status is >= 400 and not silenced.\n\n        Args:\n            response (aiohttp.ClientResponse): The response to handle.\n\n        Raises:\n               aiohttp.ClientResponseError: When the response status is >= 400.\n        \"\"\"\n        response.raise_for_status()\n\n    async def _handle_response(self, response):\n        \"\"\"\n        Handle the aiohttp response by writing it to disk and calculating digests\n\n        Args:\n            response (aiohttp.ClientResponse): The response to handle.\n\n        Returns:\n             DownloadResult: Contains information about the result. See the DownloadResult docs for\n                 more information.\n        \"\"\"\n        if self.headers_ready_callback:\n            await self.headers_ready_callback(response.headers)\n        while True:\n            chunk = await response.content.read(1048576)  # 1 megabyte\n            if not chunk:\n                await self.finalize()\n                break  # the download is done\n            await self.handle_data(chunk)\n        return DownloadResult(\n            path=self.path,\n            artifact_attributes=self.artifact_attributes,\n            url=self.url,\n            headers=response.headers,\n        )\n\n    async def run(self, extra_data=None):\n        \"\"\"\n        Run the downloader with concurrency restriction and retry logic.\n\n        This method acquires `self.semaphore` before calling the actual download implementation\n        contained in `_run()`. This ensures that the semaphore stays acquired even as the `backoff`\n        wrapper around `_run()`, handles backoff-and-retry logic.\n\n        Args:\n            extra_data (dict): Extra data passed to the downloader.\n\n        Returns:\n            :class:`~pulpcore.plugin.download.DownloadResult` from `_run()`.\n\n        \"\"\"\n        retryable_errors = (\n            aiohttp.ClientConnectorSSLError,\n            aiohttp.ClientConnectorError,\n            aiohttp.ClientOSError,\n            aiohttp.ClientPayloadError,\n            aiohttp.ClientResponseError,\n            aiohttp.ServerDisconnectedError,\n            TimeoutError,\n            TimeoutException,\n            DigestValidationError,\n            SizeValidationError,\n        )\n\n        async with self.semaphore:\n\n            @backoff.on_exception(\n                backoff.expo,\n                retryable_errors,\n                max_tries=self.max_retries + 1,\n                giveup=http_giveup_handler,\n            )\n            async def download_wrapper():\n                self._ensure_no_broken_file()\n                try:\n                    return await self._run(extra_data=extra_data)\n                except asyncio.TimeoutError:\n                    raise TimeoutException(self.url)\n\n            return await download_wrapper()\n\n    async def _run(self, extra_data=None):\n        \"\"\"\n        Download, validate, and compute digests on the `url`. This is a coroutine.\n\n        This method is externally wrapped with backoff-and-retry behavior for some errors.\n        It retries with exponential backoff some number of times before allowing a final\n        exception to be raised.\n\n        This method provides the same return object type and documented in\n        :meth:`~pulpcore.plugin.download.BaseDownloader._run`.\n\n        Args:\n            extra_data (dict): Extra data passed by the downloader.\n        \"\"\"\n        if self.download_throttler:\n            await self.download_throttler.acquire()\n        async with self.session.get(\n            self.url, proxy=self.proxy, proxy_auth=self.proxy_auth, auth=self.auth\n        ) as response:\n            self.raise_for_status(response)\n            to_return = await self._handle_response(response)\n            await response.release()\n        if self._close_session_on_finalize:\n            await self.session.close()\n        return to_return\n\n    def _ensure_no_broken_file(self):\n        \"\"\"Upon retry reset writer back to None to get a fresh file.\"\"\"\n        if self._writer is not None:\n            self._writer.delete = True\n            self._writer.close()\n            self._writer = None\n", "path": "pulpcore/download/http.py"}], "after_files": [{"content": "import logging\n\nimport aiohttp\nimport asyncio\nimport backoff\n\nfrom .base import BaseDownloader, DownloadResult\nfrom pulpcore.exceptions import (\n    DigestValidationError,\n    SizeValidationError,\n    TimeoutException,\n)\n\n\nlog = logging.getLogger(__name__)\n\n\nlogging.getLogger(\"backoff\").addHandler(logging.StreamHandler())\n\n\ndef http_giveup_handler(exc):\n    \"\"\"\n    Inspect a raised exception and determine if we should give up.\n\n    Do not give up when the error is one of the following:\n\n        HTTP 429 - Too Many Requests\n        HTTP 5xx - Server errors\n        Socket timeout\n        TCP disconnect\n        Client SSL Error\n\n    Based on the AWS and Google Cloud guidelines:\n        https://docs.aws.amazon.com/general/latest/gr/api-retries.html\n        https://cloud.google.com/storage/docs/retry-strategy\n\n    Args:\n        exc (Exception): The exception to inspect\n\n    Returns:\n        True if the download should give up, False otherwise\n    \"\"\"\n    if isinstance(exc, aiohttp.ClientResponseError):\n        server_error = 500 <= exc.code < 600\n        too_many_requests = exc.code == 429\n        return not server_error and not too_many_requests\n\n    # any other type of error (pre-filtered by the backoff decorator) shouldn't be fatal\n    return False\n\n\nclass HttpDownloader(BaseDownloader):\n    \"\"\"\n    An HTTP/HTTPS Downloader built on `aiohttp`.\n\n    This downloader downloads data from one `url` and is not reused.\n\n    The downloader optionally takes a session argument, which is an `aiohttp.ClientSession`. This\n    allows many downloaders to share one `aiohttp.ClientSession` which provides a connection pool,\n    connection reuse, and keep-alives across multiple downloaders. When creating many downloaders,\n    have one session shared by all of your `HttpDownloader` objects.\n\n    A session is optional; if omitted, one session will be created, used for this downloader, and\n    then closed when the download is complete. A session that is passed in will not be closed when\n    the download is complete.\n\n    If a session is not provided, the one created by HttpDownloader uses non-default timing values.\n    Specifically, the \"total\" timeout is set to None and the \"sock_connect\" and \"sock_read\" are both\n    5 minutes. For more info on these settings, see the aiohttp docs:\n    http://aiohttp.readthedocs.io/en/stable/client_quickstart.html#timeouts Behaviorally, it should\n    allow for an active download to be arbitrarily long, while still detecting dead or closed\n    sessions even when TCPKeepAlive is disabled.\n\n    If a session is not provided, the one created will force TCP connection closure after each\n    request. This is done for compatibility reasons due to various issues related to session\n    continuation implementation in various servers.\n\n    `aiohttp.ClientSession` objects allows you to configure options that will apply to all\n    downloaders using that session such as auth, timeouts, headers, etc. For more info on these\n    options see the `aiohttp.ClientSession` docs for more information:\n    http://aiohttp.readthedocs.io/en/stable/client_reference.html#aiohttp.ClientSession\n\n    The `aiohttp.ClientSession` can additionally be configured for SSL configuration by passing in a\n    `aiohttp.TCPConnector`. For information on configuring either server or client certificate based\n    identity verification, see the aiohttp documentation:\n    http://aiohttp.readthedocs.io/en/stable/client.html#ssl-control-for-tcp-sockets\n\n    For more information on `aiohttp.BasicAuth` objects, see their docs:\n    http://aiohttp.readthedocs.io/en/stable/client_reference.html#aiohttp.BasicAuth\n\n    Synchronous Download::\n\n        downloader = HttpDownloader('http://example.com/')\n        result = downloader.fetch()\n\n    Parallel Download::\n\n        download_coroutines = [\n            HttpDownloader('http://example.com/').run(),\n            HttpDownloader('http://pulpproject.org/').run(),\n        ]\n\n        loop = asyncio.get_event_loop()\n        done, not_done = loop.run_until_complete(asyncio.wait(download_coroutines))\n\n        for task in done:\n            try:\n                task.result()  # This is a DownloadResult\n            except Exception as error:\n                pass  # fatal exceptions are raised by result()\n\n    The HTTPDownloaders contain automatic retry logic if the server responds with HTTP 429 response.\n    The coroutine will automatically retry 10 times with exponential backoff before allowing a\n    final exception to be raised.\n\n    Attributes:\n        session (aiohttp.ClientSession): The session to be used by the downloader.\n        auth (aiohttp.BasicAuth): An object that represents HTTP Basic Authorization or None\n        proxy (str): An optional proxy URL or None\n        proxy_auth (aiohttp.BasicAuth): An optional object that represents proxy HTTP Basic\n            Authorization or None\n        headers_ready_callback (callable): An optional callback that accepts a single dictionary\n            as its argument. The callback will be called when the response headers are\n            available. The dictionary passed has the header names as the keys and header values\n            as its values. e.g. `{'Transfer-Encoding': 'chunked'}`. This can also be None.\n\n    This downloader also has all of the attributes of\n    :class:`~pulpcore.plugin.download.BaseDownloader`\n    \"\"\"\n\n    def __init__(\n        self,\n        url,\n        session=None,\n        auth=None,\n        proxy=None,\n        proxy_auth=None,\n        headers_ready_callback=None,\n        headers=None,\n        throttler=None,\n        max_retries=0,\n        **kwargs,\n    ):\n        \"\"\"\n        Args:\n            url (str): The url to download.\n            session (aiohttp.ClientSession): The session to be used by the downloader. (optional) If\n                not specified it will open the session and close it\n            auth (aiohttp.BasicAuth): An object that represents HTTP Basic Authorization (optional)\n            proxy (str): An optional proxy URL.\n            proxy_auth (aiohttp.BasicAuth): An optional object that represents proxy HTTP Basic\n                Authorization.\n            headers_ready_callback (callable): An optional callback that accepts a single dictionary\n                as its argument. The callback will be called when the response headers are\n                available. The dictionary passed has the header names as the keys and header values\n                as its values. e.g. `{'Transfer-Encoding': 'chunked'}`\n            headers (dict): Headers to be submitted with the request.\n            throttler (asyncio_throttle.Throttler): Throttler for asyncio.\n            max_retries (int): The maximum number of times to retry a download upon failure.\n            kwargs (dict): This accepts the parameters of\n                :class:`~pulpcore.plugin.download.BaseDownloader`.\n        \"\"\"\n        if session:\n            self.session = session\n            self._close_session_on_finalize = False\n        else:\n            timeout = aiohttp.ClientTimeout(total=None, sock_connect=600, sock_read=600)\n            conn = aiohttp.TCPConnector({\"force_close\": True})\n            self.session = aiohttp.ClientSession(connector=conn, timeout=timeout, headers=headers)\n            self._close_session_on_finalize = True\n        self.auth = auth\n        self.proxy = proxy\n        self.proxy_auth = proxy_auth\n        self.headers_ready_callback = headers_ready_callback\n        self.download_throttler = throttler\n        self.max_retries = max_retries\n        super().__init__(url, **kwargs)\n\n    def raise_for_status(self, response):\n        \"\"\"\n        Raise error if aiohttp response status is >= 400 and not silenced.\n\n        Args:\n            response (aiohttp.ClientResponse): The response to handle.\n\n        Raises:\n               aiohttp.ClientResponseError: When the response status is >= 400.\n        \"\"\"\n        response.raise_for_status()\n\n    async def _handle_response(self, response):\n        \"\"\"\n        Handle the aiohttp response by writing it to disk and calculating digests\n\n        Args:\n            response (aiohttp.ClientResponse): The response to handle.\n\n        Returns:\n             DownloadResult: Contains information about the result. See the DownloadResult docs for\n                 more information.\n        \"\"\"\n        if self.headers_ready_callback:\n            await self.headers_ready_callback(response.headers)\n        while True:\n            chunk = await response.content.read(1048576)  # 1 megabyte\n            if not chunk:\n                await self.finalize()\n                break  # the download is done\n            await self.handle_data(chunk)\n        return DownloadResult(\n            path=self.path,\n            artifact_attributes=self.artifact_attributes,\n            url=self.url,\n            headers=response.headers,\n        )\n\n    async def run(self, extra_data=None):\n        \"\"\"\n        Run the downloader with concurrency restriction and retry logic.\n\n        This method acquires `self.semaphore` before calling the actual download implementation\n        contained in `_run()`. This ensures that the semaphore stays acquired even as the `backoff`\n        wrapper around `_run()`, handles backoff-and-retry logic.\n\n        Args:\n            extra_data (dict): Extra data passed to the downloader.\n\n        Returns:\n            :class:`~pulpcore.plugin.download.DownloadResult` from `_run()`.\n\n        \"\"\"\n        retryable_errors = (\n            aiohttp.ClientConnectorSSLError,\n            aiohttp.ClientConnectorError,\n            aiohttp.ClientOSError,\n            aiohttp.ClientPayloadError,\n            aiohttp.ClientResponseError,\n            aiohttp.ServerDisconnectedError,\n            TimeoutError,\n            TimeoutException,\n            DigestValidationError,\n            SizeValidationError,\n        )\n\n        async with self.semaphore:\n\n            @backoff.on_exception(\n                backoff.expo,\n                retryable_errors,\n                max_tries=self.max_retries + 1,\n                giveup=http_giveup_handler,\n            )\n            async def download_wrapper():\n                self._ensure_no_broken_file()\n                try:\n                    return await self._run(extra_data=extra_data)\n                except asyncio.TimeoutError:\n                    raise TimeoutException(self.url)\n                except aiohttp.ClientHttpProxyError as e:\n                    log.error(\n                        \"Proxy {!r} rejected connection request during a request to \"\n                        \"{!r}, status={}, message={!r}\".format(\n                            e.request_info.real_url,\n                            e.request_info.url,\n                            e.status,\n                            e.message,\n                        )\n                    )\n                    raise e\n\n            return await download_wrapper()\n\n    async def _run(self, extra_data=None):\n        \"\"\"\n        Download, validate, and compute digests on the `url`. This is a coroutine.\n\n        This method is externally wrapped with backoff-and-retry behavior for some errors.\n        It retries with exponential backoff some number of times before allowing a final\n        exception to be raised.\n\n        This method provides the same return object type and documented in\n        :meth:`~pulpcore.plugin.download.BaseDownloader._run`.\n\n        Args:\n            extra_data (dict): Extra data passed by the downloader.\n        \"\"\"\n        if self.download_throttler:\n            await self.download_throttler.acquire()\n        async with self.session.get(\n            self.url, proxy=self.proxy, proxy_auth=self.proxy_auth, auth=self.auth\n        ) as response:\n            self.raise_for_status(response)\n            to_return = await self._handle_response(response)\n            await response.release()\n        if self._close_session_on_finalize:\n            await self.session.close()\n        return to_return\n\n    def _ensure_no_broken_file(self):\n        \"\"\"Upon retry reset writer back to None to get a fresh file.\"\"\"\n        if self._writer is not None:\n            self._writer.delete = True\n            self._writer.close()\n            self._writer = None\n", "path": "pulpcore/download/http.py"}]}