We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
BaseDownloader.fetch passes coroutine to asyncio.wait which is forbidden in python 3.11
Python 3.8 deprecated passing coroutines to `asyncio.wait` and Python 3.11 will now [raise an error](https://github.com/python/cpython/blob/a6313d78f21f79ca64dedd38e637509dc530a1b6/Lib/asyncio/tasks.py#L414C13-L414C13). This causes the BaseDownloader.fetch call to fail on Python 3.11 https://github.com/pulp/pulpcore/blob/9dbcc8810f97f53297a933df2e1b74cdc324a8ea/pulpcore/download/base.py#L185 .

Python provides the solution in the error message:  "Passing coroutines is forbidden, use tasks explicitly."

I believe this can be fixed by explicitly converting the coroutine to a task using asyncio's `create_task`

--- 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/base.py`
Content:
```
1 from gettext import gettext as _
2 
3 import asyncio
4 from collections import namedtuple
5 import logging
6 import os
7 import tempfile
8 from urllib.parse import urlsplit
9 
10 from pulpcore.app import pulp_hashlib
11 from pulpcore.app.models import Artifact
12 from pulpcore.exceptions import (
13     DigestValidationError,
14     SizeValidationError,
15     TimeoutException,
16     UnsupportedDigestValidationError,
17 )
18 
19 
20 log = logging.getLogger(__name__)
21 
22 
23 DownloadResult = namedtuple("DownloadResult", ["url", "artifact_attributes", "path", "headers"])
24 """
25 Args:
26     url (str): The url corresponding with the download.
27     path (str): The absolute path to the saved file
28     artifact_attributes (dict): Contains keys corresponding with
29         :class:`~pulpcore.plugin.models.Artifact` fields. This includes the computed digest values
30         along with size information.
31     headers (aiohttp.multidict.MultiDict): HTTP response headers. The keys are header names. The
32         values are header content. None when not using the HttpDownloader or sublclass.
33 """
34 
35 
36 class BaseDownloader:
37     """
38     The base class of all downloaders, providing digest calculation, validation, and file handling.
39 
40     This is an abstract class and is meant to be subclassed. Subclasses are required to implement
41     the :meth:`~pulpcore.plugin.download.BaseDownloader.run` method and do two things:
42 
43         1. Pass all downloaded data to
44            :meth:`~pulpcore.plugin.download.BaseDownloader.handle_data` and schedule it.
45 
46         2. Schedule :meth:`~pulpcore.plugin.download.BaseDownloader.finalize` after all data has
47            been delivered to :meth:`~pulpcore.plugin.download.BaseDownloader.handle_data`.
48 
49     Passing all downloaded data the into
50     :meth:`~pulpcore.plugin.download.BaseDownloader.handle_data` allows the file digests to
51     be computed while data is written to disk. The digests computed are required if the download is
52     to be saved as an :class:`~pulpcore.plugin.models.Artifact` which avoids having to re-read the
53     data later.
54 
55     The :meth:`~pulpcore.plugin.download.BaseDownloader.handle_data` method by default
56     writes to a random file in the current working directory.
57 
58     The call to :meth:`~pulpcore.plugin.download.BaseDownloader.finalize` ensures that all
59     data written to the file-like object is quiesced to disk before the file-like object has
60     `close()` called on it.
61 
62     Attributes:
63         url (str): The url to download.
64         expected_digests (dict): Keyed on the algorithm name provided by hashlib and stores the
65             value of the expected digest. e.g. {'md5': '912ec803b2ce49e4a541068d495ab570'}
66         expected_size (int): The number of bytes the download is expected to have.
67         path (str): The full path to the file containing the downloaded data.
68     """
69 
70     def __init__(
71         self,
72         url,
73         expected_digests=None,
74         expected_size=None,
75         semaphore=None,
76         *args,
77         **kwargs,
78     ):
79         """
80         Create a BaseDownloader object. This is expected to be called by all subclasses.
81 
82         Args:
83             url (str): The url to download.
84             expected_digests (dict): Keyed on the algorithm name provided by hashlib and stores the
85                 value of the expected digest. e.g. {'md5': '912ec803b2ce49e4a541068d495ab570'}
86             expected_size (int): The number of bytes the download is expected to have.
87             semaphore (asyncio.Semaphore): A semaphore the downloader must acquire before running.
88                 Useful for limiting the number of outstanding downloaders in various ways.
89         """
90 
91         self.url = url
92         self._writer = None
93         self.path = None
94         self.expected_digests = expected_digests
95         self.expected_size = expected_size
96         if semaphore:
97             self.semaphore = semaphore
98         else:
99             self.semaphore = asyncio.Semaphore()  # This will always be acquired
100         self._digests = {}
101         self._size = 0
102         if self.expected_digests:
103             if not set(self.expected_digests).intersection(set(Artifact.DIGEST_FIELDS)):
104                 raise UnsupportedDigestValidationError(
105                     _(
106                         "Content at the URL '{}' does not contain at least one trusted hasher which"
107                         " is specified in the 'ALLOWED_CONTENT_CHECKSUMS' setting ({}). The"
108                         " downloader expected one of the following hashers: {}"
109                     ).format(self.url, Artifact.DIGEST_FIELDS, set(self.expected_digests))
110                 )
111 
112     def _ensure_writer_has_open_file(self):
113         """
114         Create a temporary file on demand.
115 
116         Create a temporary file when it's actually used,
117         allowing plugin writers to instantiate many downloaders in memory.
118         """
119         if not self._writer:
120             filename = urlsplit(self.url).path.split("/")[-1]
121             # linux allows any character except NUL or / in a filename and has a length limit of
122             # 255. Making it urlencoding-aware would be nice, but not critical, because urlencoded
123             # paths should be OK
124             is_legal_filename = filename and (len(filename) <= 243)  # 255 - prefix length
125             # if the filename isn't legal then we just fall back to no suffix (random name)
126             suffix = "-" + filename if is_legal_filename else None
127             # write the file to the current working directory with a random prefix and the
128             # desired suffix. we always want the random prefix as it is possible to download
129             # the same filename from two different URLs, and the files may not be the same.
130             self._writer = tempfile.NamedTemporaryFile(dir=".", suffix=suffix, delete=False)
131             self.path = self._writer.name
132             self._digests = {n: pulp_hashlib.new(n) for n in Artifact.DIGEST_FIELDS}
133             self._size = 0
134 
135     async def handle_data(self, data):
136         """
137         A coroutine that writes data to the file object and compute its digests.
138 
139         All subclassed downloaders are expected to pass all data downloaded to this method. Similar
140         to the hashlib docstring, repeated calls are equivalent to a single call with
141         the concatenation of all the arguments: m.handle_data(a); m.handle_data(b) is equivalent to
142         m.handle_data(a+b).
143 
144         Args:
145             data (bytes): The data to be handled by the downloader.
146         """
147         self._ensure_writer_has_open_file()
148         self._writer.write(data)
149         self._record_size_and_digests_for_data(data)
150 
151     async def finalize(self):
152         """
153         A coroutine to flush downloaded data, close the file writer, and validate the data.
154 
155         All subclasses are required to call this method after all data has been passed to
156         :meth:`~pulpcore.plugin.download.BaseDownloader.handle_data`.
157 
158         Raises:
159             :class:`~pulpcore.exceptions.DigestValidationError`: When any of the ``expected_digest``
160                 values don't match the digest of the data passed to
161                 :meth:`~pulpcore.plugin.download.BaseDownloader.handle_data`.
162             :class:`~pulpcore.exceptions.SizeValidationError`: When the ``expected_size`` value
163                 doesn't match the size of the data passed to
164                 :meth:`~pulpcore.plugin.download.BaseDownloader.handle_data`.
165         """
166         self._ensure_writer_has_open_file()
167         self._writer.flush()
168         os.fsync(self._writer.fileno())
169         self._writer.close()
170         self._writer = None
171         self.validate_digests()
172         self.validate_size()
173         log.debug(f"Downloaded file from {self.url}")
174 
175     def fetch(self):
176         """
177         Run the download synchronously and return the `DownloadResult`.
178 
179         Returns:
180             :class:`~pulpcore.plugin.download.DownloadResult`
181 
182         Raises:
183             Exception: Any fatal exception emitted during downloading
184         """
185         done, _ = asyncio.get_event_loop().run_until_complete(asyncio.wait([self.run()]))
186         return done.pop().result()
187 
188     def _record_size_and_digests_for_data(self, data):
189         """
190         Record the size and digest for an available chunk of data.
191 
192         Args:
193             data (bytes): The data to have its size and digest values recorded.
194         """
195         for algorithm in self._digests.values():
196             algorithm.update(data)
197         self._size += len(data)
198 
199     @property
200     def artifact_attributes(self):
201         """
202         A property that returns a dictionary with size and digest information. The keys of this
203         dictionary correspond with :class:`~pulpcore.plugin.models.Artifact` fields.
204         """
205         attributes = {"size": self._size}
206         for algorithm in self._digests:
207             attributes[algorithm] = self._digests[algorithm].hexdigest()
208         return attributes
209 
210     def validate_digests(self):
211         """
212         Validate all digests validate if ``expected_digests`` is set
213 
214         Raises:
215             :class:`~pulpcore.exceptions.DigestValidationError`: When any of the ``expected_digest``
216                 values don't match the digest of the data passed to
217                 :meth:`~pulpcore.plugin.download.BaseDownloader.handle_data`.
218         """
219         if self.expected_digests:
220             for algorithm, expected_digest in self.expected_digests.items():
221                 actual_digest = self._digests[algorithm].hexdigest()
222                 if actual_digest != expected_digest:
223                     raise DigestValidationError(actual_digest, expected_digest, url=self.url)
224 
225     def validate_size(self):
226         """
227         Validate the size if ``expected_size`` is set
228 
229         Raises:
230             :class:`~pulpcore.exceptions.SizeValidationError`: When the ``expected_size`` value
231                 doesn't match the size of the data passed to
232                 :meth:`~pulpcore.plugin.download.BaseDownloader.handle_data`.
233         """
234         if self.expected_size:
235             actual_size = self._size
236             expected_size = self.expected_size
237             if actual_size != expected_size:
238                 raise SizeValidationError(actual_size, expected_size, url=self.url)
239 
240     async def run(self, extra_data=None):
241         """
242         Run the downloader with concurrency restriction.
243 
244         This method acquires `self.semaphore` before calling the actual download implementation
245         contained in `_run()`. This ensures that the semaphore stays acquired even as the `backoff`
246         decorator on `_run()`, handles backoff-and-retry logic.
247 
248         Args:
249             extra_data (dict): Extra data passed to the downloader.
250 
251         Returns:
252             :class:`~pulpcore.plugin.download.DownloadResult` from `_run()`.
253 
254         """
255         async with self.semaphore:
256             try:
257                 return await self._run(extra_data=extra_data)
258             except asyncio.TimeoutError:
259                 raise TimeoutException(self.url)
260 
261     async def _run(self, extra_data=None):
262         """
263         Run the downloader.
264 
265         This is a coroutine that asyncio can schedule to complete downloading. Subclasses are
266         required to implement this method and do two things:
267 
268         1. Pass all downloaded data to
269            :meth:`~pulpcore.plugin.download.BaseDownloader.handle_data`.
270 
271         2. Call :meth:`~pulpcore.plugin.download.BaseDownloader.finalize` after all data has
272            been delivered to :meth:`~pulpcore.plugin.download.BaseDownloader.handle_data`.
273 
274         It is also expected that the subclass implementation return a
275         :class:`~pulpcore.plugin.download.DownloadResult` object. The
276         ``artifact_attributes`` value of the
277         :class:`~pulpcore.plugin.download.DownloadResult` is usually set to the
278         :attr:`~pulpcore.plugin.download.BaseDownloader.artifact_attributes` property value.
279 
280         This method is called from :meth:`~pulpcore.plugin.download.BaseDownloader.run` which
281         handles concurrency restriction. Thus, by the time this method is called, the download can
282         occur without violating the concurrency restriction.
283 
284         Args:
285             extra_data (dict): Extra data passed to the downloader.
286 
287         Returns:
288             :class:`~pulpcore.plugin.download.DownloadResult`
289 
290         Raises:
291             Validation errors could be emitted when subclassed implementations call
292             :meth:`~pulpcore.plugin.download.BaseDownloader.finalize`.
293         """
294         raise NotImplementedError("Subclasses must define a _run() method that returns a coroutine")
295
```
--- 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/base.py b/pulpcore/download/base.py
--- a/pulpcore/download/base.py
+++ b/pulpcore/download/base.py
@@ -182,8 +182,8 @@
         Raises:
             Exception: Any fatal exception emitted during downloading
         """
-        done, _ = asyncio.get_event_loop().run_until_complete(asyncio.wait([self.run()]))
-        return done.pop().result()
+        result = asyncio.get_event_loop().run_until_complete(self.run())
+        return result
 
     def _record_size_and_digests_for_data(self, data):
         """

{"golden_diff": "diff --git a/pulpcore/download/base.py b/pulpcore/download/base.py\n--- a/pulpcore/download/base.py\n+++ b/pulpcore/download/base.py\n@@ -182,8 +182,8 @@\n         Raises:\n             Exception: Any fatal exception emitted during downloading\n         \"\"\"\n-        done, _ = asyncio.get_event_loop().run_until_complete(asyncio.wait([self.run()]))\n-        return done.pop().result()\n+        result = asyncio.get_event_loop().run_until_complete(self.run())\n+        return result\n \n     def _record_size_and_digests_for_data(self, data):\n         \"\"\"\n", "issue": "BaseDownloader.fetch passes coroutine to asyncio.wait which is forbidden in python 3.11\nPython 3.8 deprecated passing coroutines to `asyncio.wait` and Python 3.11 will now [raise an error](https://github.com/python/cpython/blob/a6313d78f21f79ca64dedd38e637509dc530a1b6/Lib/asyncio/tasks.py#L414C13-L414C13). This causes the BaseDownloader.fetch call to fail on Python 3.11 https://github.com/pulp/pulpcore/blob/9dbcc8810f97f53297a933df2e1b74cdc324a8ea/pulpcore/download/base.py#L185 .\r\n\r\nPython provides the solution in the error message:  \"Passing coroutines is forbidden, use tasks explicitly.\"\r\n\r\nI believe this can be fixed by explicitly converting the coroutine to a task using asyncio's `create_task`\n", "before_files": [{"content": "from gettext import gettext as _\n\nimport asyncio\nfrom collections import namedtuple\nimport logging\nimport os\nimport tempfile\nfrom urllib.parse import urlsplit\n\nfrom pulpcore.app import pulp_hashlib\nfrom pulpcore.app.models import Artifact\nfrom pulpcore.exceptions import (\n    DigestValidationError,\n    SizeValidationError,\n    TimeoutException,\n    UnsupportedDigestValidationError,\n)\n\n\nlog = logging.getLogger(__name__)\n\n\nDownloadResult = namedtuple(\"DownloadResult\", [\"url\", \"artifact_attributes\", \"path\", \"headers\"])\n\"\"\"\nArgs:\n    url (str): The url corresponding with the download.\n    path (str): The absolute path to the saved file\n    artifact_attributes (dict): Contains keys corresponding with\n        :class:`~pulpcore.plugin.models.Artifact` fields. This includes the computed digest values\n        along with size information.\n    headers (aiohttp.multidict.MultiDict): HTTP response headers. The keys are header names. The\n        values are header content. None when not using the HttpDownloader or sublclass.\n\"\"\"\n\n\nclass BaseDownloader:\n    \"\"\"\n    The base class of all downloaders, providing digest calculation, validation, and file handling.\n\n    This is an abstract class and is meant to be subclassed. Subclasses are required to implement\n    the :meth:`~pulpcore.plugin.download.BaseDownloader.run` method and do two things:\n\n        1. Pass all downloaded data to\n           :meth:`~pulpcore.plugin.download.BaseDownloader.handle_data` and schedule it.\n\n        2. Schedule :meth:`~pulpcore.plugin.download.BaseDownloader.finalize` after all data has\n           been delivered to :meth:`~pulpcore.plugin.download.BaseDownloader.handle_data`.\n\n    Passing all downloaded data the into\n    :meth:`~pulpcore.plugin.download.BaseDownloader.handle_data` allows the file digests to\n    be computed while data is written to disk. The digests computed are required if the download is\n    to be saved as an :class:`~pulpcore.plugin.models.Artifact` which avoids having to re-read the\n    data later.\n\n    The :meth:`~pulpcore.plugin.download.BaseDownloader.handle_data` method by default\n    writes to a random file in the current working directory.\n\n    The call to :meth:`~pulpcore.plugin.download.BaseDownloader.finalize` ensures that all\n    data written to the file-like object is quiesced to disk before the file-like object has\n    `close()` called on it.\n\n    Attributes:\n        url (str): The url to download.\n        expected_digests (dict): Keyed on the algorithm name provided by hashlib and stores the\n            value of the expected digest. e.g. {'md5': '912ec803b2ce49e4a541068d495ab570'}\n        expected_size (int): The number of bytes the download is expected to have.\n        path (str): The full path to the file containing the downloaded data.\n    \"\"\"\n\n    def __init__(\n        self,\n        url,\n        expected_digests=None,\n        expected_size=None,\n        semaphore=None,\n        *args,\n        **kwargs,\n    ):\n        \"\"\"\n        Create a BaseDownloader object. This is expected to be called by all subclasses.\n\n        Args:\n            url (str): The url to download.\n            expected_digests (dict): Keyed on the algorithm name provided by hashlib and stores the\n                value of the expected digest. e.g. {'md5': '912ec803b2ce49e4a541068d495ab570'}\n            expected_size (int): The number of bytes the download is expected to have.\n            semaphore (asyncio.Semaphore): A semaphore the downloader must acquire before running.\n                Useful for limiting the number of outstanding downloaders in various ways.\n        \"\"\"\n\n        self.url = url\n        self._writer = None\n        self.path = None\n        self.expected_digests = expected_digests\n        self.expected_size = expected_size\n        if semaphore:\n            self.semaphore = semaphore\n        else:\n            self.semaphore = asyncio.Semaphore()  # This will always be acquired\n        self._digests = {}\n        self._size = 0\n        if self.expected_digests:\n            if not set(self.expected_digests).intersection(set(Artifact.DIGEST_FIELDS)):\n                raise UnsupportedDigestValidationError(\n                    _(\n                        \"Content at the URL '{}' does not contain at least one trusted hasher which\"\n                        \" is specified in the 'ALLOWED_CONTENT_CHECKSUMS' setting ({}). The\"\n                        \" downloader expected one of the following hashers: {}\"\n                    ).format(self.url, Artifact.DIGEST_FIELDS, set(self.expected_digests))\n                )\n\n    def _ensure_writer_has_open_file(self):\n        \"\"\"\n        Create a temporary file on demand.\n\n        Create a temporary file when it's actually used,\n        allowing plugin writers to instantiate many downloaders in memory.\n        \"\"\"\n        if not self._writer:\n            filename = urlsplit(self.url).path.split(\"/\")[-1]\n            # linux allows any character except NUL or / in a filename and has a length limit of\n            # 255. Making it urlencoding-aware would be nice, but not critical, because urlencoded\n            # paths should be OK\n            is_legal_filename = filename and (len(filename) <= 243)  # 255 - prefix length\n            # if the filename isn't legal then we just fall back to no suffix (random name)\n            suffix = \"-\" + filename if is_legal_filename else None\n            # write the file to the current working directory with a random prefix and the\n            # desired suffix. we always want the random prefix as it is possible to download\n            # the same filename from two different URLs, and the files may not be the same.\n            self._writer = tempfile.NamedTemporaryFile(dir=\".\", suffix=suffix, delete=False)\n            self.path = self._writer.name\n            self._digests = {n: pulp_hashlib.new(n) for n in Artifact.DIGEST_FIELDS}\n            self._size = 0\n\n    async def handle_data(self, data):\n        \"\"\"\n        A coroutine that writes data to the file object and compute its digests.\n\n        All subclassed downloaders are expected to pass all data downloaded to this method. Similar\n        to the hashlib docstring, repeated calls are equivalent to a single call with\n        the concatenation of all the arguments: m.handle_data(a); m.handle_data(b) is equivalent to\n        m.handle_data(a+b).\n\n        Args:\n            data (bytes): The data to be handled by the downloader.\n        \"\"\"\n        self._ensure_writer_has_open_file()\n        self._writer.write(data)\n        self._record_size_and_digests_for_data(data)\n\n    async def finalize(self):\n        \"\"\"\n        A coroutine to flush downloaded data, close the file writer, and validate the data.\n\n        All subclasses are required to call this method after all data has been passed to\n        :meth:`~pulpcore.plugin.download.BaseDownloader.handle_data`.\n\n        Raises:\n            :class:`~pulpcore.exceptions.DigestValidationError`: When any of the ``expected_digest``\n                values don't match the digest of the data passed to\n                :meth:`~pulpcore.plugin.download.BaseDownloader.handle_data`.\n            :class:`~pulpcore.exceptions.SizeValidationError`: When the ``expected_size`` value\n                doesn't match the size of the data passed to\n                :meth:`~pulpcore.plugin.download.BaseDownloader.handle_data`.\n        \"\"\"\n        self._ensure_writer_has_open_file()\n        self._writer.flush()\n        os.fsync(self._writer.fileno())\n        self._writer.close()\n        self._writer = None\n        self.validate_digests()\n        self.validate_size()\n        log.debug(f\"Downloaded file from {self.url}\")\n\n    def fetch(self):\n        \"\"\"\n        Run the download synchronously and return the `DownloadResult`.\n\n        Returns:\n            :class:`~pulpcore.plugin.download.DownloadResult`\n\n        Raises:\n            Exception: Any fatal exception emitted during downloading\n        \"\"\"\n        done, _ = asyncio.get_event_loop().run_until_complete(asyncio.wait([self.run()]))\n        return done.pop().result()\n\n    def _record_size_and_digests_for_data(self, data):\n        \"\"\"\n        Record the size and digest for an available chunk of data.\n\n        Args:\n            data (bytes): The data to have its size and digest values recorded.\n        \"\"\"\n        for algorithm in self._digests.values():\n            algorithm.update(data)\n        self._size += len(data)\n\n    @property\n    def artifact_attributes(self):\n        \"\"\"\n        A property that returns a dictionary with size and digest information. The keys of this\n        dictionary correspond with :class:`~pulpcore.plugin.models.Artifact` fields.\n        \"\"\"\n        attributes = {\"size\": self._size}\n        for algorithm in self._digests:\n            attributes[algorithm] = self._digests[algorithm].hexdigest()\n        return attributes\n\n    def validate_digests(self):\n        \"\"\"\n        Validate all digests validate if ``expected_digests`` is set\n\n        Raises:\n            :class:`~pulpcore.exceptions.DigestValidationError`: When any of the ``expected_digest``\n                values don't match the digest of the data passed to\n                :meth:`~pulpcore.plugin.download.BaseDownloader.handle_data`.\n        \"\"\"\n        if self.expected_digests:\n            for algorithm, expected_digest in self.expected_digests.items():\n                actual_digest = self._digests[algorithm].hexdigest()\n                if actual_digest != expected_digest:\n                    raise DigestValidationError(actual_digest, expected_digest, url=self.url)\n\n    def validate_size(self):\n        \"\"\"\n        Validate the size if ``expected_size`` is set\n\n        Raises:\n            :class:`~pulpcore.exceptions.SizeValidationError`: When the ``expected_size`` value\n                doesn't match the size of the data passed to\n                :meth:`~pulpcore.plugin.download.BaseDownloader.handle_data`.\n        \"\"\"\n        if self.expected_size:\n            actual_size = self._size\n            expected_size = self.expected_size\n            if actual_size != expected_size:\n                raise SizeValidationError(actual_size, expected_size, url=self.url)\n\n    async def run(self, extra_data=None):\n        \"\"\"\n        Run the downloader with concurrency restriction.\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        decorator on `_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        async with self.semaphore:\n            try:\n                return await self._run(extra_data=extra_data)\n            except asyncio.TimeoutError:\n                raise TimeoutException(self.url)\n\n    async def _run(self, extra_data=None):\n        \"\"\"\n        Run the downloader.\n\n        This is a coroutine that asyncio can schedule to complete downloading. Subclasses are\n        required to implement this method and do two things:\n\n        1. Pass all downloaded data to\n           :meth:`~pulpcore.plugin.download.BaseDownloader.handle_data`.\n\n        2. Call :meth:`~pulpcore.plugin.download.BaseDownloader.finalize` after all data has\n           been delivered to :meth:`~pulpcore.plugin.download.BaseDownloader.handle_data`.\n\n        It is also expected that the subclass implementation return a\n        :class:`~pulpcore.plugin.download.DownloadResult` object. The\n        ``artifact_attributes`` value of the\n        :class:`~pulpcore.plugin.download.DownloadResult` is usually set to the\n        :attr:`~pulpcore.plugin.download.BaseDownloader.artifact_attributes` property value.\n\n        This method is called from :meth:`~pulpcore.plugin.download.BaseDownloader.run` which\n        handles concurrency restriction. Thus, by the time this method is called, the download can\n        occur without violating the concurrency restriction.\n\n        Args:\n            extra_data (dict): Extra data passed to the downloader.\n\n        Returns:\n            :class:`~pulpcore.plugin.download.DownloadResult`\n\n        Raises:\n            Validation errors could be emitted when subclassed implementations call\n            :meth:`~pulpcore.plugin.download.BaseDownloader.finalize`.\n        \"\"\"\n        raise NotImplementedError(\"Subclasses must define a _run() method that returns a coroutine\")\n", "path": "pulpcore/download/base.py"}], "after_files": [{"content": "from gettext import gettext as _\n\nimport asyncio\nfrom collections import namedtuple\nimport logging\nimport os\nimport tempfile\nfrom urllib.parse import urlsplit\n\nfrom pulpcore.app import pulp_hashlib\nfrom pulpcore.app.models import Artifact\nfrom pulpcore.exceptions import (\n    DigestValidationError,\n    SizeValidationError,\n    TimeoutException,\n    UnsupportedDigestValidationError,\n)\n\n\nlog = logging.getLogger(__name__)\n\n\nDownloadResult = namedtuple(\"DownloadResult\", [\"url\", \"artifact_attributes\", \"path\", \"headers\"])\n\"\"\"\nArgs:\n    url (str): The url corresponding with the download.\n    path (str): The absolute path to the saved file\n    artifact_attributes (dict): Contains keys corresponding with\n        :class:`~pulpcore.plugin.models.Artifact` fields. This includes the computed digest values\n        along with size information.\n    headers (aiohttp.multidict.MultiDict): HTTP response headers. The keys are header names. The\n        values are header content. None when not using the HttpDownloader or sublclass.\n\"\"\"\n\n\nclass BaseDownloader:\n    \"\"\"\n    The base class of all downloaders, providing digest calculation, validation, and file handling.\n\n    This is an abstract class and is meant to be subclassed. Subclasses are required to implement\n    the :meth:`~pulpcore.plugin.download.BaseDownloader.run` method and do two things:\n\n        1. Pass all downloaded data to\n           :meth:`~pulpcore.plugin.download.BaseDownloader.handle_data` and schedule it.\n\n        2. Schedule :meth:`~pulpcore.plugin.download.BaseDownloader.finalize` after all data has\n           been delivered to :meth:`~pulpcore.plugin.download.BaseDownloader.handle_data`.\n\n    Passing all downloaded data the into\n    :meth:`~pulpcore.plugin.download.BaseDownloader.handle_data` allows the file digests to\n    be computed while data is written to disk. The digests computed are required if the download is\n    to be saved as an :class:`~pulpcore.plugin.models.Artifact` which avoids having to re-read the\n    data later.\n\n    The :meth:`~pulpcore.plugin.download.BaseDownloader.handle_data` method by default\n    writes to a random file in the current working directory.\n\n    The call to :meth:`~pulpcore.plugin.download.BaseDownloader.finalize` ensures that all\n    data written to the file-like object is quiesced to disk before the file-like object has\n    `close()` called on it.\n\n    Attributes:\n        url (str): The url to download.\n        expected_digests (dict): Keyed on the algorithm name provided by hashlib and stores the\n            value of the expected digest. e.g. {'md5': '912ec803b2ce49e4a541068d495ab570'}\n        expected_size (int): The number of bytes the download is expected to have.\n        path (str): The full path to the file containing the downloaded data.\n    \"\"\"\n\n    def __init__(\n        self,\n        url,\n        expected_digests=None,\n        expected_size=None,\n        semaphore=None,\n        *args,\n        **kwargs,\n    ):\n        \"\"\"\n        Create a BaseDownloader object. This is expected to be called by all subclasses.\n\n        Args:\n            url (str): The url to download.\n            expected_digests (dict): Keyed on the algorithm name provided by hashlib and stores the\n                value of the expected digest. e.g. {'md5': '912ec803b2ce49e4a541068d495ab570'}\n            expected_size (int): The number of bytes the download is expected to have.\n            semaphore (asyncio.Semaphore): A semaphore the downloader must acquire before running.\n                Useful for limiting the number of outstanding downloaders in various ways.\n        \"\"\"\n\n        self.url = url\n        self._writer = None\n        self.path = None\n        self.expected_digests = expected_digests\n        self.expected_size = expected_size\n        if semaphore:\n            self.semaphore = semaphore\n        else:\n            self.semaphore = asyncio.Semaphore()  # This will always be acquired\n        self._digests = {}\n        self._size = 0\n        if self.expected_digests:\n            if not set(self.expected_digests).intersection(set(Artifact.DIGEST_FIELDS)):\n                raise UnsupportedDigestValidationError(\n                    _(\n                        \"Content at the URL '{}' does not contain at least one trusted hasher which\"\n                        \" is specified in the 'ALLOWED_CONTENT_CHECKSUMS' setting ({}). The\"\n                        \" downloader expected one of the following hashers: {}\"\n                    ).format(self.url, Artifact.DIGEST_FIELDS, set(self.expected_digests))\n                )\n\n    def _ensure_writer_has_open_file(self):\n        \"\"\"\n        Create a temporary file on demand.\n\n        Create a temporary file when it's actually used,\n        allowing plugin writers to instantiate many downloaders in memory.\n        \"\"\"\n        if not self._writer:\n            filename = urlsplit(self.url).path.split(\"/\")[-1]\n            # linux allows any character except NUL or / in a filename and has a length limit of\n            # 255. Making it urlencoding-aware would be nice, but not critical, because urlencoded\n            # paths should be OK\n            is_legal_filename = filename and (len(filename) <= 243)  # 255 - prefix length\n            # if the filename isn't legal then we just fall back to no suffix (random name)\n            suffix = \"-\" + filename if is_legal_filename else None\n            # write the file to the current working directory with a random prefix and the\n            # desired suffix. we always want the random prefix as it is possible to download\n            # the same filename from two different URLs, and the files may not be the same.\n            self._writer = tempfile.NamedTemporaryFile(dir=\".\", suffix=suffix, delete=False)\n            self.path = self._writer.name\n            self._digests = {n: pulp_hashlib.new(n) for n in Artifact.DIGEST_FIELDS}\n            self._size = 0\n\n    async def handle_data(self, data):\n        \"\"\"\n        A coroutine that writes data to the file object and compute its digests.\n\n        All subclassed downloaders are expected to pass all data downloaded to this method. Similar\n        to the hashlib docstring, repeated calls are equivalent to a single call with\n        the concatenation of all the arguments: m.handle_data(a); m.handle_data(b) is equivalent to\n        m.handle_data(a+b).\n\n        Args:\n            data (bytes): The data to be handled by the downloader.\n        \"\"\"\n        self._ensure_writer_has_open_file()\n        self._writer.write(data)\n        self._record_size_and_digests_for_data(data)\n\n    async def finalize(self):\n        \"\"\"\n        A coroutine to flush downloaded data, close the file writer, and validate the data.\n\n        All subclasses are required to call this method after all data has been passed to\n        :meth:`~pulpcore.plugin.download.BaseDownloader.handle_data`.\n\n        Raises:\n            :class:`~pulpcore.exceptions.DigestValidationError`: When any of the ``expected_digest``\n                values don't match the digest of the data passed to\n                :meth:`~pulpcore.plugin.download.BaseDownloader.handle_data`.\n            :class:`~pulpcore.exceptions.SizeValidationError`: When the ``expected_size`` value\n                doesn't match the size of the data passed to\n                :meth:`~pulpcore.plugin.download.BaseDownloader.handle_data`.\n        \"\"\"\n        self._ensure_writer_has_open_file()\n        self._writer.flush()\n        os.fsync(self._writer.fileno())\n        self._writer.close()\n        self._writer = None\n        self.validate_digests()\n        self.validate_size()\n        log.debug(f\"Downloaded file from {self.url}\")\n\n    def fetch(self):\n        \"\"\"\n        Run the download synchronously and return the `DownloadResult`.\n\n        Returns:\n            :class:`~pulpcore.plugin.download.DownloadResult`\n\n        Raises:\n            Exception: Any fatal exception emitted during downloading\n        \"\"\"\n        result = asyncio.get_event_loop().run_until_complete(self.run())\n        return result\n\n    def _record_size_and_digests_for_data(self, data):\n        \"\"\"\n        Record the size and digest for an available chunk of data.\n\n        Args:\n            data (bytes): The data to have its size and digest values recorded.\n        \"\"\"\n        for algorithm in self._digests.values():\n            algorithm.update(data)\n        self._size += len(data)\n\n    @property\n    def artifact_attributes(self):\n        \"\"\"\n        A property that returns a dictionary with size and digest information. The keys of this\n        dictionary correspond with :class:`~pulpcore.plugin.models.Artifact` fields.\n        \"\"\"\n        attributes = {\"size\": self._size}\n        for algorithm in self._digests:\n            attributes[algorithm] = self._digests[algorithm].hexdigest()\n        return attributes\n\n    def validate_digests(self):\n        \"\"\"\n        Validate all digests validate if ``expected_digests`` is set\n\n        Raises:\n            :class:`~pulpcore.exceptions.DigestValidationError`: When any of the ``expected_digest``\n                values don't match the digest of the data passed to\n                :meth:`~pulpcore.plugin.download.BaseDownloader.handle_data`.\n        \"\"\"\n        if self.expected_digests:\n            for algorithm, expected_digest in self.expected_digests.items():\n                actual_digest = self._digests[algorithm].hexdigest()\n                if actual_digest != expected_digest:\n                    raise DigestValidationError(actual_digest, expected_digest, url=self.url)\n\n    def validate_size(self):\n        \"\"\"\n        Validate the size if ``expected_size`` is set\n\n        Raises:\n            :class:`~pulpcore.exceptions.SizeValidationError`: When the ``expected_size`` value\n                doesn't match the size of the data passed to\n                :meth:`~pulpcore.plugin.download.BaseDownloader.handle_data`.\n        \"\"\"\n        if self.expected_size:\n            actual_size = self._size\n            expected_size = self.expected_size\n            if actual_size != expected_size:\n                raise SizeValidationError(actual_size, expected_size, url=self.url)\n\n    async def run(self, extra_data=None):\n        \"\"\"\n        Run the downloader with concurrency restriction.\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        decorator on `_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        async with self.semaphore:\n            try:\n                return await self._run(extra_data=extra_data)\n            except asyncio.TimeoutError:\n                raise TimeoutException(self.url)\n\n    async def _run(self, extra_data=None):\n        \"\"\"\n        Run the downloader.\n\n        This is a coroutine that asyncio can schedule to complete downloading. Subclasses are\n        required to implement this method and do two things:\n\n        1. Pass all downloaded data to\n           :meth:`~pulpcore.plugin.download.BaseDownloader.handle_data`.\n\n        2. Call :meth:`~pulpcore.plugin.download.BaseDownloader.finalize` after all data has\n           been delivered to :meth:`~pulpcore.plugin.download.BaseDownloader.handle_data`.\n\n        It is also expected that the subclass implementation return a\n        :class:`~pulpcore.plugin.download.DownloadResult` object. The\n        ``artifact_attributes`` value of the\n        :class:`~pulpcore.plugin.download.DownloadResult` is usually set to the\n        :attr:`~pulpcore.plugin.download.BaseDownloader.artifact_attributes` property value.\n\n        This method is called from :meth:`~pulpcore.plugin.download.BaseDownloader.run` which\n        handles concurrency restriction. Thus, by the time this method is called, the download can\n        occur without violating the concurrency restriction.\n\n        Args:\n            extra_data (dict): Extra data passed to the downloader.\n\n        Returns:\n            :class:`~pulpcore.plugin.download.DownloadResult`\n\n        Raises:\n            Validation errors could be emitted when subclassed implementations call\n            :meth:`~pulpcore.plugin.download.BaseDownloader.finalize`.\n        \"\"\"\n        raise NotImplementedError(\"Subclasses must define a _run() method that returns a coroutine\")\n", "path": "pulpcore/download/base.py"}]}