Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# Copyright 2017 Google Inc.
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
15"""Virtual bases classes for uploading media via Google APIs.
17Supported here are:
19* simple (media) uploads
20* multipart uploads that contain both metadata and a small file as payload
21* resumable uploads (with metadata as well)
22"""
24import http.client
25import json
26import os
27import random
28import re
29import sys
30import urllib.parse
31from xml.etree import ElementTree
33from google import resumable_media
34from google.resumable_media import _helpers, common
36_CONTENT_TYPE_HEADER = "content-type"
37_CONTENT_RANGE_TEMPLATE = "bytes {:d}-{:d}/{:d}"
38_RANGE_UNKNOWN_TEMPLATE = "bytes {:d}-{:d}/*"
39_EMPTY_RANGE_TEMPLATE = "bytes */{:d}"
40_BOUNDARY_WIDTH = len(str(sys.maxsize - 1))
41_BOUNDARY_FORMAT = "==============={{:0{:d}d}}==".format(_BOUNDARY_WIDTH)
42_MULTIPART_SEP = b"--"
43_CRLF = b"\r\n"
44_MULTIPART_BEGIN = b"\r\ncontent-type: application/json; charset=UTF-8\r\n\r\n"
45_RELATED_HEADER = b'multipart/related; boundary="'
46_BYTES_RANGE_RE = re.compile(r"bytes=0-(?P<end_byte>\d+)", flags=re.IGNORECASE)
47_STREAM_ERROR_TEMPLATE = (
48 "Bytes stream is in unexpected state. "
49 "The local stream has had {:d} bytes read from it while "
50 "{:d} bytes have already been updated (they should match)."
51)
52_STREAM_READ_PAST_TEMPLATE = (
53 "{:d} bytes have been read from the stream, which exceeds the expected total {:d}."
54)
55_DELETE = "DELETE"
56_POST = "POST"
57_PUT = "PUT"
58_UPLOAD_CHECKSUM_MISMATCH_MESSAGE = (
59 "The computed ``{}`` checksum, ``{}``, and the checksum reported by the "
60 "remote host, ``{}``, did not match."
61)
62_UPLOAD_METADATA_NO_APPROPRIATE_CHECKSUM_MESSAGE = (
63 "Response metadata had no ``{}`` value; checksum could not be validated."
64)
65_UPLOAD_HEADER_NO_APPROPRIATE_CHECKSUM_MESSAGE = (
66 "Response headers had no ``{}`` value; checksum could not be validated."
67)
68_MPU_INITIATE_QUERY = "?uploads"
69_MPU_PART_QUERY_TEMPLATE = "?partNumber={part}&uploadId={upload_id}"
70_S3_COMPAT_XML_NAMESPACE = "{http://s3.amazonaws.com/doc/2006-03-01/}"
71_UPLOAD_ID_NODE = "UploadId"
72_MPU_FINAL_QUERY_TEMPLATE = "?uploadId={upload_id}"
75class UploadBase(object):
76 """Base class for upload helpers.
78 Defines core shared behavior across different upload types.
80 Args:
81 upload_url (str): The URL where the content will be uploaded.
82 headers (Optional[Mapping[str, str]]): Extra headers that should
83 be sent with the request, e.g. headers for encrypted data.
85 Attributes:
86 upload_url (str): The URL where the content will be uploaded.
87 """
89 def __init__(self, upload_url, headers=None):
90 self.upload_url = upload_url
91 if headers is None:
92 headers = {}
93 self._headers = headers
94 self._finished = False
95 self._retry_strategy = common.RetryStrategy()
97 @property
98 def finished(self):
99 """bool: Flag indicating if the upload has completed."""
100 return self._finished
102 def _process_response(self, response):
103 """Process the response from an HTTP request.
105 This is everything that must be done after a request that doesn't
106 require network I/O (or other I/O). This is based on the `sans-I/O`_
107 philosophy.
109 Args:
110 response (object): The HTTP response object.
112 Raises:
113 ~google.resumable_media.common.InvalidResponse: If the status
114 code is not 200.
116 .. _sans-I/O: https://sans-io.readthedocs.io/
117 """
118 # Tombstone the current upload so it cannot be used again (in either
119 # failure or success).
120 self._finished = True
121 _helpers.require_status_code(response, (http.client.OK,), self._get_status_code)
123 @staticmethod
124 def _get_status_code(response):
125 """Access the status code from an HTTP response.
127 Args:
128 response (object): The HTTP response object.
130 Raises:
131 NotImplementedError: Always, since virtual.
132 """
133 raise NotImplementedError("This implementation is virtual.")
135 @staticmethod
136 def _get_headers(response):
137 """Access the headers from an HTTP response.
139 Args:
140 response (object): The HTTP response object.
142 Raises:
143 NotImplementedError: Always, since virtual.
144 """
145 raise NotImplementedError("This implementation is virtual.")
147 @staticmethod
148 def _get_body(response):
149 """Access the response body from an HTTP response.
151 Args:
152 response (object): The HTTP response object.
154 Raises:
155 NotImplementedError: Always, since virtual.
156 """
157 raise NotImplementedError("This implementation is virtual.")
160class SimpleUpload(UploadBase):
161 """Upload a resource to a Google API.
163 A **simple** media upload sends no metadata and completes the upload
164 in a single request.
166 Args:
167 upload_url (str): The URL where the content will be uploaded.
168 headers (Optional[Mapping[str, str]]): Extra headers that should
169 be sent with the request, e.g. headers for encrypted data.
171 Attributes:
172 upload_url (str): The URL where the content will be uploaded.
173 """
175 def _prepare_request(self, data, content_type):
176 """Prepare the contents of an HTTP request.
178 This is everything that must be done before a request that doesn't
179 require network I/O (or other I/O). This is based on the `sans-I/O`_
180 philosophy.
182 .. note:
184 This method will be used only once, so ``headers`` will be
185 mutated by having a new key added to it.
187 Args:
188 data (bytes): The resource content to be uploaded.
189 content_type (str): The content type for the request.
191 Returns:
192 Tuple[str, str, bytes, Mapping[str, str]]: The quadruple
194 * HTTP verb for the request (always POST)
195 * the URL for the request
196 * the body of the request
197 * headers for the request
199 Raises:
200 ValueError: If the current upload has already finished.
201 TypeError: If ``data`` isn't bytes.
203 .. _sans-I/O: https://sans-io.readthedocs.io/
204 """
205 if self.finished:
206 raise ValueError("An upload can only be used once.")
208 if not isinstance(data, bytes):
209 raise TypeError("`data` must be bytes, received", type(data))
210 self._headers[_CONTENT_TYPE_HEADER] = content_type
211 return _POST, self.upload_url, data, self._headers
213 def transmit(self, transport, data, content_type, timeout=None):
214 """Transmit the resource to be uploaded.
216 Args:
217 transport (object): An object which can make authenticated
218 requests.
219 data (bytes): The resource content to be uploaded.
220 content_type (str): The content type of the resource, e.g. a JPEG
221 image has content type ``image/jpeg``.
222 timeout (Optional[Union[float, Tuple[float, float]]]):
223 The number of seconds to wait for the server response.
224 Depending on the retry strategy, a request may be repeated
225 several times using the same timeout each time.
227 Can also be passed as a tuple (connect_timeout, read_timeout).
228 See :meth:`requests.Session.request` documentation for details.
230 Raises:
231 NotImplementedError: Always, since virtual.
232 """
233 raise NotImplementedError("This implementation is virtual.")
236class MultipartUpload(UploadBase):
237 """Upload a resource with metadata to a Google API.
239 A **multipart** upload sends both metadata and the resource in a single
240 (multipart) request.
242 Args:
243 upload_url (str): The URL where the content will be uploaded.
244 headers (Optional[Mapping[str, str]]): Extra headers that should
245 be sent with the request, e.g. headers for encrypted data.
246 checksum (Optional([str])): The type of checksum to compute to verify
247 the integrity of the object. The request metadata will be amended
248 to include the computed value. Using this option will override a
249 manually-set checksum value. Supported values are "md5", "crc32c"
250 and None. The default is None.
252 Attributes:
253 upload_url (str): The URL where the content will be uploaded.
254 """
256 def __init__(self, upload_url, headers=None, checksum=None):
257 super(MultipartUpload, self).__init__(upload_url, headers=headers)
258 self._checksum_type = checksum
260 def _prepare_request(self, data, metadata, content_type):
261 """Prepare the contents of an HTTP request.
263 This is everything that must be done before a request that doesn't
264 require network I/O (or other I/O). This is based on the `sans-I/O`_
265 philosophy.
267 .. note:
269 This method will be used only once, so ``headers`` will be
270 mutated by having a new key added to it.
272 Args:
273 data (bytes): The resource content to be uploaded.
274 metadata (Mapping[str, str]): The resource metadata, such as an
275 ACL list.
276 content_type (str): The content type of the resource, e.g. a JPEG
277 image has content type ``image/jpeg``.
279 Returns:
280 Tuple[str, str, bytes, Mapping[str, str]]: The quadruple
282 * HTTP verb for the request (always POST)
283 * the URL for the request
284 * the body of the request
285 * headers for the request
287 Raises:
288 ValueError: If the current upload has already finished.
289 TypeError: If ``data`` isn't bytes.
291 .. _sans-I/O: https://sans-io.readthedocs.io/
292 """
293 if self.finished:
294 raise ValueError("An upload can only be used once.")
296 if not isinstance(data, bytes):
297 raise TypeError("`data` must be bytes, received", type(data))
299 checksum_object = _helpers._get_checksum_object(self._checksum_type)
300 if checksum_object is not None:
301 checksum_object.update(data)
302 actual_checksum = _helpers.prepare_checksum_digest(checksum_object.digest())
303 metadata_key = _helpers._get_metadata_key(self._checksum_type)
304 metadata[metadata_key] = actual_checksum
306 content, multipart_boundary = construct_multipart_request(
307 data, metadata, content_type
308 )
309 multipart_content_type = _RELATED_HEADER + multipart_boundary + b'"'
310 self._headers[_CONTENT_TYPE_HEADER] = multipart_content_type
312 return _POST, self.upload_url, content, self._headers
314 def transmit(self, transport, data, metadata, content_type, timeout=None):
315 """Transmit the resource to be uploaded.
317 Args:
318 transport (object): An object which can make authenticated
319 requests.
320 data (bytes): The resource content to be uploaded.
321 metadata (Mapping[str, str]): The resource metadata, such as an
322 ACL list.
323 content_type (str): The content type of the resource, e.g. a JPEG
324 image has content type ``image/jpeg``.
325 timeout (Optional[Union[float, Tuple[float, float]]]):
326 The number of seconds to wait for the server response.
327 Depending on the retry strategy, a request may be repeated
328 several times using the same timeout each time.
330 Can also be passed as a tuple (connect_timeout, read_timeout).
331 See :meth:`requests.Session.request` documentation for details.
333 Raises:
334 NotImplementedError: Always, since virtual.
335 """
336 raise NotImplementedError("This implementation is virtual.")
339class ResumableUpload(UploadBase):
340 """Initiate and fulfill a resumable upload to a Google API.
342 A **resumable** upload sends an initial request with the resource metadata
343 and then gets assigned an upload ID / upload URL to send bytes to.
344 Using the upload URL, the upload is then done in chunks (determined by
345 the user) until all bytes have been uploaded.
347 Args:
348 upload_url (str): The URL where the resumable upload will be initiated.
349 chunk_size (int): The size of each chunk used to upload the resource.
350 headers (Optional[Mapping[str, str]]): Extra headers that should
351 be sent with every request.
352 checksum (Optional([str])): The type of checksum to compute to verify
353 the integrity of the object. After the upload is complete, the
354 server-computed checksum of the resulting object will be read
355 and google.resumable_media.common.DataCorruption will be raised on
356 a mismatch. The corrupted file will not be deleted from the remote
357 host automatically. Supported values are "md5", "crc32c" and None.
358 The default is None.
360 Attributes:
361 upload_url (str): The URL where the content will be uploaded.
363 Raises:
364 ValueError: If ``chunk_size`` is not a multiple of
365 :data:`.UPLOAD_CHUNK_SIZE`.
366 """
368 def __init__(self, upload_url, chunk_size, checksum=None, headers=None):
369 super(ResumableUpload, self).__init__(upload_url, headers=headers)
370 if chunk_size % resumable_media.UPLOAD_CHUNK_SIZE != 0:
371 raise ValueError(
372 "{} KB must divide chunk size".format(
373 resumable_media.UPLOAD_CHUNK_SIZE / 1024
374 )
375 )
376 self._chunk_size = chunk_size
377 self._stream = None
378 self._content_type = None
379 self._bytes_uploaded = 0
380 self._bytes_checksummed = 0
381 self._checksum_type = checksum
382 self._checksum_object = None
383 self._total_bytes = None
384 self._resumable_url = None
385 self._invalid = False
387 @property
388 def invalid(self):
389 """bool: Indicates if the upload is in an invalid state.
391 This will occur if a call to :meth:`transmit_next_chunk` fails.
392 To recover from such a failure, call :meth:`recover`.
393 """
394 return self._invalid
396 @property
397 def chunk_size(self):
398 """int: The size of each chunk used to upload the resource."""
399 return self._chunk_size
401 @property
402 def resumable_url(self):
403 """Optional[str]: The URL of the in-progress resumable upload."""
404 return self._resumable_url
406 @property
407 def bytes_uploaded(self):
408 """int: Number of bytes that have been uploaded."""
409 return self._bytes_uploaded
411 @property
412 def total_bytes(self):
413 """Optional[int]: The total number of bytes to be uploaded.
415 If this upload is initiated (via :meth:`initiate`) with
416 ``stream_final=True``, this value will be populated based on the size
417 of the ``stream`` being uploaded. (By default ``stream_final=True``.)
419 If this upload is initiated with ``stream_final=False``,
420 :attr:`total_bytes` will be :data:`None` since it cannot be
421 determined from the stream.
422 """
423 return self._total_bytes
425 def _prepare_initiate_request(
426 self, stream, metadata, content_type, total_bytes=None, stream_final=True
427 ):
428 """Prepare the contents of HTTP request to initiate upload.
430 This is everything that must be done before a request that doesn't
431 require network I/O (or other I/O). This is based on the `sans-I/O`_
432 philosophy.
434 Args:
435 stream (IO[bytes]): The stream (i.e. file-like object) that will
436 be uploaded. The stream **must** be at the beginning (i.e.
437 ``stream.tell() == 0``).
438 metadata (Mapping[str, str]): The resource metadata, such as an
439 ACL list.
440 content_type (str): The content type of the resource, e.g. a JPEG
441 image has content type ``image/jpeg``.
442 total_bytes (Optional[int]): The total number of bytes to be
443 uploaded. If specified, the upload size **will not** be
444 determined from the stream (even if ``stream_final=True``).
445 stream_final (Optional[bool]): Indicates if the ``stream`` is
446 "final" (i.e. no more bytes will be added to it). In this case
447 we determine the upload size from the size of the stream. If
448 ``total_bytes`` is passed, this argument will be ignored.
450 Returns:
451 Tuple[str, str, bytes, Mapping[str, str]]: The quadruple
453 * HTTP verb for the request (always POST)
454 * the URL for the request
455 * the body of the request
456 * headers for the request
458 Raises:
459 ValueError: If the current upload has already been initiated.
460 ValueError: If ``stream`` is not at the beginning.
462 .. _sans-I/O: https://sans-io.readthedocs.io/
463 """
464 if self.resumable_url is not None:
465 raise ValueError("This upload has already been initiated.")
466 if stream.tell() != 0:
467 raise ValueError("Stream must be at beginning.")
469 self._stream = stream
470 self._content_type = content_type
472 # Signed URL requires content type set directly - not through x-upload-content-type
473 parse_result = urllib.parse.urlparse(self.upload_url)
474 parsed_query = urllib.parse.parse_qs(parse_result.query)
475 if "x-goog-signature" in parsed_query or "X-Goog-Signature" in parsed_query:
476 # Deconstruct **self._headers first so that content type defined here takes priority
477 headers = {**self._headers, _CONTENT_TYPE_HEADER: content_type}
478 else:
479 # Deconstruct **self._headers first so that content type defined here takes priority
480 headers = {
481 **self._headers,
482 _CONTENT_TYPE_HEADER: "application/json; charset=UTF-8",
483 "x-upload-content-type": content_type,
484 }
485 # Set the total bytes if possible.
486 if total_bytes is not None:
487 self._total_bytes = total_bytes
488 elif stream_final:
489 self._total_bytes = get_total_bytes(stream)
490 # Add the total bytes to the headers if set.
491 if self._total_bytes is not None:
492 content_length = "{:d}".format(self._total_bytes)
493 headers["x-upload-content-length"] = content_length
495 payload = json.dumps(metadata).encode("utf-8")
496 return _POST, self.upload_url, payload, headers
498 def _process_initiate_response(self, response):
499 """Process the response from an HTTP request that initiated upload.
501 This is everything that must be done after a request that doesn't
502 require network I/O (or other I/O). This is based on the `sans-I/O`_
503 philosophy.
505 This method takes the URL from the ``Location`` header and stores it
506 for future use. Within that URL, we assume the ``upload_id`` query
507 parameter has been included, but we do not check.
509 Args:
510 response (object): The HTTP response object (need headers).
512 .. _sans-I/O: https://sans-io.readthedocs.io/
513 """
514 _helpers.require_status_code(
515 response,
516 (http.client.OK, http.client.CREATED),
517 self._get_status_code,
518 callback=self._make_invalid,
519 )
520 self._resumable_url = _helpers.header_required(
521 response, "location", self._get_headers
522 )
524 def initiate(
525 self,
526 transport,
527 stream,
528 metadata,
529 content_type,
530 total_bytes=None,
531 stream_final=True,
532 timeout=None,
533 ):
534 """Initiate a resumable upload.
536 By default, this method assumes your ``stream`` is in a "final"
537 state ready to transmit. However, ``stream_final=False`` can be used
538 to indicate that the size of the resource is not known. This can happen
539 if bytes are being dynamically fed into ``stream``, e.g. if the stream
540 is attached to application logs.
542 If ``stream_final=False`` is used, :attr:`chunk_size` bytes will be
543 read from the stream every time :meth:`transmit_next_chunk` is called.
544 If one of those reads produces strictly fewer bites than the chunk
545 size, the upload will be concluded.
547 Args:
548 transport (object): An object which can make authenticated
549 requests.
550 stream (IO[bytes]): The stream (i.e. file-like object) that will
551 be uploaded. The stream **must** be at the beginning (i.e.
552 ``stream.tell() == 0``).
553 metadata (Mapping[str, str]): The resource metadata, such as an
554 ACL list.
555 content_type (str): The content type of the resource, e.g. a JPEG
556 image has content type ``image/jpeg``.
557 total_bytes (Optional[int]): The total number of bytes to be
558 uploaded. If specified, the upload size **will not** be
559 determined from the stream (even if ``stream_final=True``).
560 stream_final (Optional[bool]): Indicates if the ``stream`` is
561 "final" (i.e. no more bytes will be added to it). In this case
562 we determine the upload size from the size of the stream. If
563 ``total_bytes`` is passed, this argument will be ignored.
564 timeout (Optional[Union[float, Tuple[float, float]]]):
565 The number of seconds to wait for the server response.
566 Depending on the retry strategy, a request may be repeated
567 several times using the same timeout each time.
569 Can also be passed as a tuple (connect_timeout, read_timeout).
570 See :meth:`requests.Session.request` documentation for details.
572 Raises:
573 NotImplementedError: Always, since virtual.
574 """
575 raise NotImplementedError("This implementation is virtual.")
577 def _prepare_request(self):
578 """Prepare the contents of HTTP request to upload a chunk.
580 This is everything that must be done before a request that doesn't
581 require network I/O. This is based on the `sans-I/O`_ philosophy.
583 For the time being, this **does require** some form of I/O to read
584 a chunk from ``stream`` (via :func:`get_next_chunk`). However, this
585 will (almost) certainly not be network I/O.
587 Returns:
588 Tuple[str, str, bytes, Mapping[str, str]]: The quadruple
590 * HTTP verb for the request (always PUT)
591 * the URL for the request
592 * the body of the request
593 * headers for the request
595 The headers incorporate the ``_headers`` on the current instance.
597 Raises:
598 ValueError: If the current upload has finished.
599 ValueError: If the current upload is in an invalid state.
600 ValueError: If the current upload has not been initiated.
601 ValueError: If the location in the stream (i.e. ``stream.tell()``)
602 does not agree with ``bytes_uploaded``.
604 .. _sans-I/O: https://sans-io.readthedocs.io/
605 """
606 if self.finished:
607 raise ValueError("Upload has finished.")
608 if self.invalid:
609 raise ValueError(
610 "Upload is in an invalid state. To recover call `recover()`."
611 )
612 if self.resumable_url is None:
613 raise ValueError(
614 "This upload has not been initiated. Please call "
615 "initiate() before beginning to transmit chunks."
616 )
618 start_byte, payload, content_range = get_next_chunk(
619 self._stream, self._chunk_size, self._total_bytes
620 )
621 if start_byte != self.bytes_uploaded:
622 msg = _STREAM_ERROR_TEMPLATE.format(start_byte, self.bytes_uploaded)
623 raise ValueError(msg)
625 self._update_checksum(start_byte, payload)
627 headers = {
628 **self._headers,
629 _CONTENT_TYPE_HEADER: self._content_type,
630 _helpers.CONTENT_RANGE_HEADER: content_range,
631 }
632 return _PUT, self.resumable_url, payload, headers
634 def _update_checksum(self, start_byte, payload):
635 """Update the checksum with the payload if not already updated.
637 Because error recovery can result in bytes being transmitted more than
638 once, the checksum tracks the number of bytes checked in
639 self._bytes_checksummed and skips bytes that have already been summed.
640 """
641 if not self._checksum_type:
642 return
644 if not self._checksum_object:
645 self._checksum_object = _helpers._get_checksum_object(self._checksum_type)
647 if start_byte < self._bytes_checksummed:
648 offset = self._bytes_checksummed - start_byte
649 data = payload[offset:]
650 else:
651 data = payload
653 self._checksum_object.update(data)
654 self._bytes_checksummed += len(data)
656 def _make_invalid(self):
657 """Simple setter for ``invalid``.
659 This is intended to be passed along as a callback to helpers that
660 raise an exception so they can mark this instance as invalid before
661 raising.
662 """
663 self._invalid = True
665 def _process_resumable_response(self, response, bytes_sent):
666 """Process the response from an HTTP request.
668 This is everything that must be done after a request that doesn't
669 require network I/O (or other I/O). This is based on the `sans-I/O`_
670 philosophy.
672 Args:
673 response (object): The HTTP response object.
674 bytes_sent (int): The number of bytes sent in the request that
675 ``response`` was returned for.
677 Raises:
678 ~google.resumable_media.common.InvalidResponse: If the status
679 code is 308 and the ``range`` header is not of the form
680 ``bytes 0-{end}``.
681 ~google.resumable_media.common.InvalidResponse: If the status
682 code is not 200 or 308.
684 .. _sans-I/O: https://sans-io.readthedocs.io/
685 """
686 status_code = _helpers.require_status_code(
687 response,
688 (http.client.OK, http.client.PERMANENT_REDIRECT),
689 self._get_status_code,
690 callback=self._make_invalid,
691 )
692 if status_code == http.client.OK:
693 # NOTE: We use the "local" information of ``bytes_sent`` to update
694 # ``bytes_uploaded``, but do not verify this against other
695 # state. However, there may be some other information:
696 #
697 # * a ``size`` key in JSON response body
698 # * the ``total_bytes`` attribute (if set)
699 # * ``stream.tell()`` (relying on fact that ``initiate()``
700 # requires stream to be at the beginning)
701 self._bytes_uploaded = self._bytes_uploaded + bytes_sent
702 # Tombstone the current upload so it cannot be used again.
703 self._finished = True
704 # Validate the checksum. This can raise an exception on failure.
705 self._validate_checksum(response)
706 else:
707 bytes_range = _helpers.header_required(
708 response,
709 _helpers.RANGE_HEADER,
710 self._get_headers,
711 callback=self._make_invalid,
712 )
713 match = _BYTES_RANGE_RE.match(bytes_range)
714 if match is None:
715 self._make_invalid()
716 raise common.InvalidResponse(
717 response,
718 'Unexpected "range" header',
719 bytes_range,
720 'Expected to be of the form "bytes=0-{end}"',
721 )
722 self._bytes_uploaded = int(match.group("end_byte")) + 1
724 def _validate_checksum(self, response):
725 """Check the computed checksum, if any, against the recieved metadata.
727 Args:
728 response (object): The HTTP response object.
730 Raises:
731 ~google.resumable_media.common.DataCorruption: If the checksum
732 computed locally and the checksum reported by the remote host do
733 not match.
734 """
735 if self._checksum_type is None:
736 return
737 metadata_key = _helpers._get_metadata_key(self._checksum_type)
738 metadata = response.json()
739 remote_checksum = metadata.get(metadata_key)
740 if remote_checksum is None:
741 raise common.InvalidResponse(
742 response,
743 _UPLOAD_METADATA_NO_APPROPRIATE_CHECKSUM_MESSAGE.format(metadata_key),
744 self._get_headers(response),
745 )
746 local_checksum = _helpers.prepare_checksum_digest(
747 self._checksum_object.digest()
748 )
749 if local_checksum != remote_checksum:
750 raise common.DataCorruption(
751 response,
752 _UPLOAD_CHECKSUM_MISMATCH_MESSAGE.format(
753 self._checksum_type.upper(), local_checksum, remote_checksum
754 ),
755 )
757 def transmit_next_chunk(self, transport, timeout=None):
758 """Transmit the next chunk of the resource to be uploaded.
760 If the current upload was initiated with ``stream_final=False``,
761 this method will dynamically determine if the upload has completed.
762 The upload will be considered complete if the stream produces
763 fewer than :attr:`chunk_size` bytes when a chunk is read from it.
765 Args:
766 transport (object): An object which can make authenticated
767 requests.
768 timeout (Optional[Union[float, Tuple[float, float]]]):
769 The number of seconds to wait for the server response.
770 Depending on the retry strategy, a request may be repeated
771 several times using the same timeout each time.
773 Can also be passed as a tuple (connect_timeout, read_timeout).
774 See :meth:`requests.Session.request` documentation for details.
776 Raises:
777 NotImplementedError: Always, since virtual.
778 """
779 raise NotImplementedError("This implementation is virtual.")
781 def _prepare_recover_request(self):
782 """Prepare the contents of HTTP request to recover from failure.
784 This is everything that must be done before a request that doesn't
785 require network I/O. This is based on the `sans-I/O`_ philosophy.
787 We assume that the :attr:`resumable_url` is set (i.e. the only way
788 the upload can end up :attr:`invalid` is if it has been initiated.
790 Returns:
791 Tuple[str, str, NoneType, Mapping[str, str]]: The quadruple
793 * HTTP verb for the request (always PUT)
794 * the URL for the request
795 * the body of the request (always :data:`None`)
796 * headers for the request
798 The headers **do not** incorporate the ``_headers`` on the
799 current instance.
801 .. _sans-I/O: https://sans-io.readthedocs.io/
802 """
803 headers = {_helpers.CONTENT_RANGE_HEADER: "bytes */*"}
804 return _PUT, self.resumable_url, None, headers
806 def _process_recover_response(self, response):
807 """Process the response from an HTTP request to recover from failure.
809 This is everything that must be done after a request that doesn't
810 require network I/O (or other I/O). This is based on the `sans-I/O`_
811 philosophy.
813 Args:
814 response (object): The HTTP response object.
816 Raises:
817 ~google.resumable_media.common.InvalidResponse: If the status
818 code is not 308.
819 ~google.resumable_media.common.InvalidResponse: If the status
820 code is 308 and the ``range`` header is not of the form
821 ``bytes 0-{end}``.
823 .. _sans-I/O: https://sans-io.readthedocs.io/
824 """
825 _helpers.require_status_code(
826 response, (http.client.PERMANENT_REDIRECT,), self._get_status_code
827 )
828 headers = self._get_headers(response)
829 if _helpers.RANGE_HEADER in headers:
830 bytes_range = headers[_helpers.RANGE_HEADER]
831 match = _BYTES_RANGE_RE.match(bytes_range)
832 if match is None:
833 raise common.InvalidResponse(
834 response,
835 'Unexpected "range" header',
836 bytes_range,
837 'Expected to be of the form "bytes=0-{end}"',
838 )
839 self._bytes_uploaded = int(match.group("end_byte")) + 1
840 else:
841 # In this case, the upload has not "begun".
842 self._bytes_uploaded = 0
844 self._stream.seek(self._bytes_uploaded)
845 self._invalid = False
847 def recover(self, transport):
848 """Recover from a failure.
850 This method should be used when a :class:`ResumableUpload` is in an
851 :attr:`~ResumableUpload.invalid` state due to a request failure.
853 This will verify the progress with the server and make sure the
854 current upload is in a valid state before :meth:`transmit_next_chunk`
855 can be used again.
857 Args:
858 transport (object): An object which can make authenticated
859 requests.
861 Raises:
862 NotImplementedError: Always, since virtual.
863 """
864 raise NotImplementedError("This implementation is virtual.")
867class XMLMPUContainer(UploadBase):
868 """Initiate and close an upload using the XML MPU API.
870 An XML MPU sends an initial request and then receives an upload ID.
871 Using the upload ID, the upload is then done in numbered parts and the
872 parts can be uploaded concurrently.
874 In order to avoid concurrency issues with this container object, the
875 uploading of individual parts is handled separately, by XMLMPUPart objects
876 spawned from this container class. The XMLMPUPart objects are not
877 necessarily in the same process as the container, so they do not update the
878 container automatically.
880 MPUs are sometimes referred to as "Multipart Uploads", which is ambiguous
881 given the JSON multipart upload, so the abbreviation "MPU" will be used
882 throughout.
884 See: https://cloud.google.com/storage/docs/multipart-uploads
886 Args:
887 upload_url (str): The URL of the object (without query parameters). The
888 initiate, PUT, and finalization requests will all use this URL, with
889 varying query parameters.
890 filename (str): The name (path) of the file to upload.
891 headers (Optional[Mapping[str, str]]): Extra headers that should
892 be sent with every request.
894 Attributes:
895 upload_url (str): The URL where the content will be uploaded.
896 upload_id (Optional(str)): The ID of the upload from the initialization
897 response.
898 """
900 def __init__(self, upload_url, filename, headers=None, upload_id=None):
901 super().__init__(upload_url, headers=headers)
902 self._filename = filename
903 self._upload_id = upload_id
904 self._parts = {}
906 @property
907 def upload_id(self):
908 return self._upload_id
910 def register_part(self, part_number, etag):
911 """Register an uploaded part by part number and corresponding etag.
913 XMLMPUPart objects represent individual parts, and their part number
914 and etag can be registered to the container object with this method
915 and therefore incorporated in the finalize() call to finish the upload.
917 This method accepts part_number and etag, but not XMLMPUPart objects
918 themselves, to reduce the complexity involved in running XMLMPUPart
919 uploads in separate processes.
921 Args:
922 part_number (int): The part number. Parts are assembled into the
923 final uploaded object with finalize() in order of their part
924 numbers.
925 etag (str): The etag included in the server response after upload.
926 """
927 self._parts[part_number] = etag
929 def _prepare_initiate_request(self, content_type):
930 """Prepare the contents of HTTP request to initiate upload.
932 This is everything that must be done before a request that doesn't
933 require network I/O (or other I/O). This is based on the `sans-I/O`_
934 philosophy.
936 Args:
937 content_type (str): The content type of the resource, e.g. a JPEG
938 image has content type ``image/jpeg``.
940 Returns:
941 Tuple[str, str, bytes, Mapping[str, str]]: The quadruple
943 * HTTP verb for the request (always POST)
944 * the URL for the request
945 * the body of the request
946 * headers for the request
948 Raises:
949 ValueError: If the current upload has already been initiated.
951 .. _sans-I/O: https://sans-io.readthedocs.io/
952 """
953 if self.upload_id is not None:
954 raise ValueError("This upload has already been initiated.")
956 initiate_url = self.upload_url + _MPU_INITIATE_QUERY
958 headers = {
959 **self._headers,
960 _CONTENT_TYPE_HEADER: content_type,
961 }
962 return _POST, initiate_url, None, headers
964 def _process_initiate_response(self, response):
965 """Process the response from an HTTP request that initiated the upload.
967 This is everything that must be done after a request that doesn't
968 require network I/O (or other I/O). This is based on the `sans-I/O`_
969 philosophy.
971 This method takes the URL from the ``Location`` header and stores it
972 for future use. Within that URL, we assume the ``upload_id`` query
973 parameter has been included, but we do not check.
975 Args:
976 response (object): The HTTP response object.
978 Raises:
979 ~google.resumable_media.common.InvalidResponse: If the status
980 code is not 200.
982 .. _sans-I/O: https://sans-io.readthedocs.io/
983 """
984 _helpers.require_status_code(response, (http.client.OK,), self._get_status_code)
985 root = ElementTree.fromstring(response.text)
986 self._upload_id = root.find(_S3_COMPAT_XML_NAMESPACE + _UPLOAD_ID_NODE).text
988 def initiate(
989 self,
990 transport,
991 content_type,
992 timeout=None,
993 ):
994 """Initiate an MPU and record the upload ID.
996 Args:
997 transport (object): An object which can make authenticated
998 requests.
999 content_type (str): The content type of the resource, e.g. a JPEG
1000 image has content type ``image/jpeg``.
1001 timeout (Optional[Union[float, Tuple[float, float]]]):
1002 The number of seconds to wait for the server response.
1003 Depending on the retry strategy, a request may be repeated
1004 several times using the same timeout each time.
1006 Can also be passed as a tuple (connect_timeout, read_timeout).
1007 See :meth:`requests.Session.request` documentation for details.
1009 Raises:
1010 NotImplementedError: Always, since virtual.
1011 """
1012 raise NotImplementedError("This implementation is virtual.")
1014 def _prepare_finalize_request(self):
1015 """Prepare the contents of an HTTP request to finalize the upload.
1017 All of the parts must be registered before calling this method.
1019 Returns:
1020 Tuple[str, str, bytes, Mapping[str, str]]: The quadruple
1022 * HTTP verb for the request (always POST)
1023 * the URL for the request
1024 * the body of the request
1025 * headers for the request
1027 Raises:
1028 ValueError: If the upload has not been initiated.
1029 """
1030 if self.upload_id is None:
1031 raise ValueError("This upload has not yet been initiated.")
1033 final_query = _MPU_FINAL_QUERY_TEMPLATE.format(upload_id=self._upload_id)
1034 finalize_url = self.upload_url + final_query
1035 final_xml_root = ElementTree.Element("CompleteMultipartUpload")
1036 for part_number, etag in self._parts.items():
1037 part = ElementTree.SubElement(final_xml_root, "Part") # put in a loop
1038 ElementTree.SubElement(part, "PartNumber").text = str(part_number)
1039 ElementTree.SubElement(part, "ETag").text = etag
1040 payload = ElementTree.tostring(final_xml_root)
1041 return _POST, finalize_url, payload, self._headers
1043 def _process_finalize_response(self, response):
1044 """Process the response from an HTTP request that finalized the upload.
1046 This is everything that must be done after a request that doesn't
1047 require network I/O (or other I/O). This is based on the `sans-I/O`_
1048 philosophy.
1050 Args:
1051 response (object): The HTTP response object.
1053 Raises:
1054 ~google.resumable_media.common.InvalidResponse: If the status
1055 code is not 200.
1057 .. _sans-I/O: https://sans-io.readthedocs.io/
1058 """
1060 _helpers.require_status_code(response, (http.client.OK,), self._get_status_code)
1061 self._finished = True
1063 def finalize(
1064 self,
1065 transport,
1066 timeout=None,
1067 ):
1068 """Finalize an MPU request with all the parts.
1070 Args:
1071 transport (object): An object which can make authenticated
1072 requests.
1073 timeout (Optional[Union[float, Tuple[float, float]]]):
1074 The number of seconds to wait for the server response.
1075 Depending on the retry strategy, a request may be repeated
1076 several times using the same timeout each time.
1078 Can also be passed as a tuple (connect_timeout, read_timeout).
1079 See :meth:`requests.Session.request` documentation for details.
1081 Raises:
1082 NotImplementedError: Always, since virtual.
1083 """
1084 raise NotImplementedError("This implementation is virtual.")
1086 def _prepare_cancel_request(self):
1087 """Prepare the contents of an HTTP request to cancel the upload.
1089 Returns:
1090 Tuple[str, str, bytes, Mapping[str, str]]: The quadruple
1092 * HTTP verb for the request (always DELETE)
1093 * the URL for the request
1094 * the body of the request
1095 * headers for the request
1097 Raises:
1098 ValueError: If the upload has not been initiated.
1099 """
1100 if self.upload_id is None:
1101 raise ValueError("This upload has not yet been initiated.")
1103 cancel_query = _MPU_FINAL_QUERY_TEMPLATE.format(upload_id=self._upload_id)
1104 cancel_url = self.upload_url + cancel_query
1105 return _DELETE, cancel_url, None, self._headers
1107 def _process_cancel_response(self, response):
1108 """Process the response from an HTTP request that canceled the upload.
1110 This is everything that must be done after a request that doesn't
1111 require network I/O (or other I/O). This is based on the `sans-I/O`_
1112 philosophy.
1114 Args:
1115 response (object): The HTTP response object.
1117 Raises:
1118 ~google.resumable_media.common.InvalidResponse: If the status
1119 code is not 204.
1121 .. _sans-I/O: https://sans-io.readthedocs.io/
1122 """
1124 _helpers.require_status_code(
1125 response, (http.client.NO_CONTENT,), self._get_status_code
1126 )
1128 def cancel(
1129 self,
1130 transport,
1131 timeout=None,
1132 ):
1133 """Cancel an MPU request and permanently delete any uploaded parts.
1135 This cannot be undone.
1137 Args:
1138 transport (object): An object which can make authenticated
1139 requests.
1140 timeout (Optional[Union[float, Tuple[float, float]]]):
1141 The number of seconds to wait for the server response.
1142 Depending on the retry strategy, a request may be repeated
1143 several times using the same timeout each time.
1145 Can also be passed as a tuple (connect_timeout, read_timeout).
1146 See :meth:`requests.Session.request` documentation for details.
1148 Raises:
1149 NotImplementedError: Always, since virtual.
1150 """
1151 raise NotImplementedError("This implementation is virtual.")
1154class XMLMPUPart(UploadBase):
1155 """Upload a single part of an existing XML MPU container.
1157 An XML MPU sends an initial request and then receives an upload ID.
1158 Using the upload ID, the upload is then done in numbered parts and the
1159 parts can be uploaded concurrently.
1161 In order to avoid concurrency issues with the container object, the
1162 uploading of individual parts is handled separately by multiple objects
1163 of this class. Once a part is uploaded, it can be registered with the
1164 container with `container.register_part(part.part_number, part.etag)`.
1166 MPUs are sometimes referred to as "Multipart Uploads", which is ambiguous
1167 given the JSON multipart upload, so the abbreviation "MPU" will be used
1168 throughout.
1170 See: https://cloud.google.com/storage/docs/multipart-uploads
1172 Args:
1173 upload_url (str): The URL of the object (without query parameters).
1174 upload_id (str): The ID of the upload from the initialization response.
1175 filename (str): The name (path) of the file to upload.
1176 start (int): The byte index of the beginning of the part.
1177 end (int): The byte index of the end of the part.
1178 part_number (int): The part number. Part numbers will be assembled in
1179 sequential order when the container is finalized.
1180 headers (Optional[Mapping[str, str]]): Extra headers that should
1181 be sent with every request.
1182 checksum (Optional([str])): The type of checksum to compute to verify
1183 the integrity of the object. The request headers will be amended
1184 to include the computed value. Supported values are "md5", "crc32c"
1185 and None. The default is None.
1187 Attributes:
1188 upload_url (str): The URL of the object (without query parameters).
1189 upload_id (str): The ID of the upload from the initialization response.
1190 filename (str): The name (path) of the file to upload.
1191 start (int): The byte index of the beginning of the part.
1192 end (int): The byte index of the end of the part.
1193 part_number (int): The part number. Part numbers will be assembled in
1194 sequential order when the container is finalized.
1195 etag (Optional(str)): The etag returned by the service after upload.
1196 """
1198 def __init__(
1199 self,
1200 upload_url,
1201 upload_id,
1202 filename,
1203 start,
1204 end,
1205 part_number,
1206 headers=None,
1207 checksum=None,
1208 ):
1209 super().__init__(upload_url, headers=headers)
1210 self._filename = filename
1211 self._start = start
1212 self._end = end
1213 self._upload_id = upload_id
1214 self._part_number = part_number
1215 self._etag = None
1216 self._checksum_type = checksum
1217 self._checksum_object = None
1219 @property
1220 def part_number(self):
1221 return self._part_number
1223 @property
1224 def upload_id(self):
1225 return self._upload_id
1227 @property
1228 def filename(self):
1229 return self._filename
1231 @property
1232 def etag(self):
1233 return self._etag
1235 @property
1236 def start(self):
1237 return self._start
1239 @property
1240 def end(self):
1241 return self._end
1243 def _prepare_upload_request(self):
1244 """Prepare the contents of HTTP request to upload a part.
1246 This is everything that must be done before a request that doesn't
1247 require network I/O. This is based on the `sans-I/O`_ philosophy.
1249 For the time being, this **does require** some form of I/O to read
1250 a part from ``stream`` (via :func:`get_part_payload`). However, this
1251 will (almost) certainly not be network I/O.
1253 Returns:
1254 Tuple[str, str, bytes, Mapping[str, str]]: The quadruple
1256 * HTTP verb for the request (always PUT)
1257 * the URL for the request
1258 * the body of the request
1259 * headers for the request
1261 The headers incorporate the ``_headers`` on the current instance.
1263 Raises:
1264 ValueError: If the current upload has finished.
1266 .. _sans-I/O: https://sans-io.readthedocs.io/
1267 """
1268 if self.finished:
1269 raise ValueError("This part has already been uploaded.")
1271 with open(self._filename, "br") as f:
1272 f.seek(self._start)
1273 payload = f.read(self._end - self._start)
1275 self._checksum_object = _helpers._get_checksum_object(self._checksum_type)
1276 if self._checksum_object is not None:
1277 self._checksum_object.update(payload)
1279 part_query = _MPU_PART_QUERY_TEMPLATE.format(
1280 part=self._part_number, upload_id=self._upload_id
1281 )
1282 upload_url = self.upload_url + part_query
1283 return _PUT, upload_url, payload, self._headers
1285 def _process_upload_response(self, response):
1286 """Process the response from an HTTP request.
1288 This is everything that must be done after a request that doesn't
1289 require network I/O (or other I/O). This is based on the `sans-I/O`_
1290 philosophy.
1292 Args:
1293 response (object): The HTTP response object.
1295 Raises:
1296 ~google.resumable_media.common.InvalidResponse: If the status
1297 code is not 200 or the response is missing data.
1299 .. _sans-I/O: https://sans-io.readthedocs.io/
1300 """
1301 _helpers.require_status_code(
1302 response,
1303 (http.client.OK,),
1304 self._get_status_code,
1305 )
1307 self._validate_checksum(response)
1309 etag = _helpers.header_required(response, "etag", self._get_headers)
1310 self._etag = etag
1311 self._finished = True
1313 def upload(
1314 self,
1315 transport,
1316 timeout=None,
1317 ):
1318 """Upload the part.
1320 Args:
1321 transport (object): An object which can make authenticated
1322 requests.
1323 timeout (Optional[Union[float, Tuple[float, float]]]):
1324 The number of seconds to wait for the server response.
1325 Depending on the retry strategy, a request may be repeated
1326 several times using the same timeout each time.
1328 Can also be passed as a tuple (connect_timeout, read_timeout).
1329 See :meth:`requests.Session.request` documentation for details.
1331 Raises:
1332 NotImplementedError: Always, since virtual.
1333 """
1334 raise NotImplementedError("This implementation is virtual.")
1336 def _validate_checksum(self, response):
1337 """Check the computed checksum, if any, against the response headers.
1339 Args:
1340 response (object): The HTTP response object.
1342 Raises:
1343 ~google.resumable_media.common.DataCorruption: If the checksum
1344 computed locally and the checksum reported by the remote host do
1345 not match.
1346 """
1347 if self._checksum_type is None:
1348 return
1350 remote_checksum = _helpers._get_uploaded_checksum_from_headers(
1351 response, self._get_headers, self._checksum_type
1352 )
1354 if remote_checksum is None:
1355 metadata_key = _helpers._get_metadata_key(self._checksum_type)
1356 raise common.InvalidResponse(
1357 response,
1358 _UPLOAD_METADATA_NO_APPROPRIATE_CHECKSUM_MESSAGE.format(metadata_key),
1359 self._get_headers(response),
1360 )
1361 local_checksum = _helpers.prepare_checksum_digest(
1362 self._checksum_object.digest()
1363 )
1364 if local_checksum != remote_checksum:
1365 raise common.DataCorruption(
1366 response,
1367 _UPLOAD_CHECKSUM_MISMATCH_MESSAGE.format(
1368 self._checksum_type.upper(), local_checksum, remote_checksum
1369 ),
1370 )
1373def get_boundary():
1374 """Get a random boundary for a multipart request.
1376 Returns:
1377 bytes: The boundary used to separate parts of a multipart request.
1378 """
1379 random_int = random.randrange(sys.maxsize)
1380 boundary = _BOUNDARY_FORMAT.format(random_int)
1381 # NOTE: Neither % formatting nor .format() are available for byte strings
1382 # in Python 3.4, so we must use unicode strings as templates.
1383 return boundary.encode("utf-8")
1386def construct_multipart_request(data, metadata, content_type):
1387 """Construct a multipart request body.
1389 Args:
1390 data (bytes): The resource content (UTF-8 encoded as bytes)
1391 to be uploaded.
1392 metadata (Mapping[str, str]): The resource metadata, such as an
1393 ACL list.
1394 content_type (str): The content type of the resource, e.g. a JPEG
1395 image has content type ``image/jpeg``.
1397 Returns:
1398 Tuple[bytes, bytes]: The multipart request body and the boundary used
1399 between each part.
1400 """
1401 multipart_boundary = get_boundary()
1402 json_bytes = json.dumps(metadata).encode("utf-8")
1403 content_type = content_type.encode("utf-8")
1404 # Combine the two parts into a multipart payload.
1405 # NOTE: We'd prefer a bytes template but are restricted by Python 3.4.
1406 boundary_sep = _MULTIPART_SEP + multipart_boundary
1407 content = (
1408 boundary_sep
1409 + _MULTIPART_BEGIN
1410 + json_bytes
1411 + _CRLF
1412 + boundary_sep
1413 + _CRLF
1414 + b"content-type: "
1415 + content_type
1416 + _CRLF
1417 + _CRLF
1418 + data # Empty line between headers and body.
1419 + _CRLF
1420 + boundary_sep
1421 + _MULTIPART_SEP
1422 )
1424 return content, multipart_boundary
1427def get_total_bytes(stream):
1428 """Determine the total number of bytes in a stream.
1430 Args:
1431 stream (IO[bytes]): The stream (i.e. file-like object).
1433 Returns:
1434 int: The number of bytes.
1435 """
1436 current_position = stream.tell()
1437 # NOTE: ``.seek()`` **should** return the same value that ``.tell()``
1438 # returns, but in Python 2, ``file`` objects do not.
1439 stream.seek(0, os.SEEK_END)
1440 end_position = stream.tell()
1441 # Go back to the initial position.
1442 stream.seek(current_position)
1444 return end_position
1447def get_next_chunk(stream, chunk_size, total_bytes):
1448 """Get a chunk from an I/O stream.
1450 The ``stream`` may have fewer bytes remaining than ``chunk_size``
1451 so it may not always be the case that
1452 ``end_byte == start_byte + chunk_size - 1``.
1454 Args:
1455 stream (IO[bytes]): The stream (i.e. file-like object).
1456 chunk_size (int): The size of the chunk to be read from the ``stream``.
1457 total_bytes (Optional[int]): The (expected) total number of bytes
1458 in the ``stream``.
1460 Returns:
1461 Tuple[int, bytes, str]: Triple of:
1463 * the start byte index
1464 * the content in between the start and end bytes (inclusive)
1465 * content range header for the chunk (slice) that has been read
1467 Raises:
1468 ValueError: If ``total_bytes == 0`` but ``stream.read()`` yields
1469 non-empty content.
1470 ValueError: If there is no data left to consume. This corresponds
1471 exactly to the case ``end_byte < start_byte``, which can only
1472 occur if ``end_byte == start_byte - 1``.
1473 """
1474 start_byte = stream.tell()
1475 if total_bytes is not None and start_byte + chunk_size >= total_bytes > 0:
1476 payload = stream.read(total_bytes - start_byte)
1477 else:
1478 payload = stream.read(chunk_size)
1479 end_byte = stream.tell() - 1
1481 num_bytes_read = len(payload)
1482 if total_bytes is None:
1483 if num_bytes_read < chunk_size:
1484 # We now **KNOW** the total number of bytes.
1485 total_bytes = end_byte + 1
1486 elif total_bytes == 0:
1487 # NOTE: We also expect ``start_byte == 0`` here but don't check
1488 # because ``_prepare_initiate_request()`` requires the
1489 # stream to be at the beginning.
1490 if num_bytes_read != 0:
1491 raise ValueError(
1492 "Stream specified as empty, but produced non-empty content."
1493 )
1494 else:
1495 if num_bytes_read == 0:
1496 raise ValueError(
1497 "Stream is already exhausted. There is no content remaining."
1498 )
1500 content_range = get_content_range(start_byte, end_byte, total_bytes)
1501 return start_byte, payload, content_range
1504def get_content_range(start_byte, end_byte, total_bytes):
1505 """Convert start, end and total into content range header.
1507 If ``total_bytes`` is not known, uses "bytes {start}-{end}/*".
1508 If we are dealing with an empty range (i.e. ``end_byte < start_byte``)
1509 then "bytes */{total}" is used.
1511 This function **ASSUMES** that if the size is not known, the caller will
1512 not also pass an empty range.
1514 Args:
1515 start_byte (int): The start (inclusive) of the byte range.
1516 end_byte (int): The end (inclusive) of the byte range.
1517 total_bytes (Optional[int]): The number of bytes in the byte
1518 range (if known).
1520 Returns:
1521 str: The content range header.
1522 """
1523 if total_bytes is None:
1524 return _RANGE_UNKNOWN_TEMPLATE.format(start_byte, end_byte)
1525 elif end_byte < start_byte:
1526 return _EMPTY_RANGE_TEMPLATE.format(total_bytes)
1527 else:
1528 return _CONTENT_RANGE_TEMPLATE.format(start_byte, end_byte, total_bytes)