We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Retried downloads are not validating properly
When retrying a partial download, the size and digests need to be reset to validate.

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