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"""Support for resumable uploads.
17Also supported here are simple (media) uploads and multipart
18uploads that contain both metadata and a small file as payload.
19"""
21from google.resumable_media import _upload
22from google.resumable_media.requests import _request_helpers
25class SimpleUpload(_request_helpers.RequestsMixin, _upload.SimpleUpload):
26 """Upload a resource to a Google API.
28 A **simple** media upload sends no metadata and completes the upload
29 in a single request.
31 Args:
32 upload_url (str): The URL where the content will be uploaded.
33 headers (Optional[Mapping[str, str]]): Extra headers that should
34 be sent with the request, e.g. headers for encrypted data.
36 Attributes:
37 upload_url (str): The URL where the content will be uploaded.
38 """
40 def transmit(
41 self,
42 transport,
43 data,
44 content_type,
45 timeout=(
46 _request_helpers._DEFAULT_CONNECT_TIMEOUT,
47 _request_helpers._DEFAULT_READ_TIMEOUT,
48 ),
49 ):
50 """Transmit the resource to be uploaded.
52 Args:
53 transport (~requests.Session): A ``requests`` object which can
54 make authenticated requests.
55 data (bytes): The resource content to be uploaded.
56 content_type (str): The content type of the resource, e.g. a JPEG
57 image has content type ``image/jpeg``.
58 timeout (Optional[Union[float, Tuple[float, float]]]):
59 The number of seconds to wait for the server response.
60 Depending on the retry strategy, a request may be repeated
61 several times using the same timeout each time.
63 Can also be passed as a tuple (connect_timeout, read_timeout).
64 See :meth:`requests.Session.request` documentation for details.
66 Returns:
67 ~requests.Response: The HTTP response returned by ``transport``.
68 """
69 method, url, payload, headers = self._prepare_request(data, content_type)
71 # Wrap the request business logic in a function to be retried.
72 def retriable_request():
73 result = transport.request(
74 method, url, data=payload, headers=headers, timeout=timeout
75 )
77 self._process_response(result)
79 return result
81 return _request_helpers.wait_and_retry(
82 retriable_request, self._get_status_code, self._retry_strategy
83 )
86class MultipartUpload(_request_helpers.RequestsMixin, _upload.MultipartUpload):
87 """Upload a resource with metadata to a Google API.
89 A **multipart** upload sends both metadata and the resource in a single
90 (multipart) request.
92 Args:
93 upload_url (str): The URL where the content will be uploaded.
94 headers (Optional[Mapping[str, str]]): Extra headers that should
95 be sent with the request, e.g. headers for encrypted data.
96 checksum Optional([str]): The type of checksum to compute to verify
97 the integrity of the object. The request metadata will be amended
98 to include the computed value. Using this option will override a
99 manually-set checksum value. Supported values are "md5",
100 "crc32c" and None. The default is None.
102 Attributes:
103 upload_url (str): The URL where the content will be uploaded.
104 """
106 def transmit(
107 self,
108 transport,
109 data,
110 metadata,
111 content_type,
112 timeout=(
113 _request_helpers._DEFAULT_CONNECT_TIMEOUT,
114 _request_helpers._DEFAULT_READ_TIMEOUT,
115 ),
116 ):
117 """Transmit the resource to be uploaded.
119 Args:
120 transport (~requests.Session): A ``requests`` object which can
121 make authenticated requests.
122 data (bytes): The resource content to be uploaded.
123 metadata (Mapping[str, str]): The resource metadata, such as an
124 ACL list.
125 content_type (str): The content type of the resource, e.g. a JPEG
126 image has content type ``image/jpeg``.
127 timeout (Optional[Union[float, Tuple[float, float]]]):
128 The number of seconds to wait for the server response.
129 Depending on the retry strategy, a request may be repeated
130 several times using the same timeout each time.
132 Can also be passed as a tuple (connect_timeout, read_timeout).
133 See :meth:`requests.Session.request` documentation for details.
135 Returns:
136 ~requests.Response: The HTTP response returned by ``transport``.
137 """
138 method, url, payload, headers = self._prepare_request(
139 data, metadata, content_type
140 )
142 # Wrap the request business logic in a function to be retried.
143 def retriable_request():
144 result = transport.request(
145 method, url, data=payload, headers=headers, timeout=timeout
146 )
148 self._process_response(result)
150 return result
152 return _request_helpers.wait_and_retry(
153 retriable_request, self._get_status_code, self._retry_strategy
154 )
157class ResumableUpload(_request_helpers.RequestsMixin, _upload.ResumableUpload):
158 """Initiate and fulfill a resumable upload to a Google API.
160 A **resumable** upload sends an initial request with the resource metadata
161 and then gets assigned an upload ID / upload URL to send bytes to.
162 Using the upload URL, the upload is then done in chunks (determined by
163 the user) until all bytes have been uploaded.
165 When constructing a resumable upload, only the resumable upload URL and
166 the chunk size are required:
168 .. testsetup:: resumable-constructor
170 bucket = 'bucket-foo'
172 .. doctest:: resumable-constructor
174 >>> from google.resumable_media.requests import ResumableUpload
175 >>>
176 >>> url_template = (
177 ... 'https://www.googleapis.com/upload/storage/v1/b/{bucket}/o?'
178 ... 'uploadType=resumable')
179 >>> upload_url = url_template.format(bucket=bucket)
180 >>>
181 >>> chunk_size = 3 * 1024 * 1024 # 3MB
182 >>> upload = ResumableUpload(upload_url, chunk_size)
184 When initiating an upload (via :meth:`initiate`), the caller is expected
185 to pass the resource being uploaded as a file-like ``stream``. If the size
186 of the resource is explicitly known, it can be passed in directly:
188 .. testsetup:: resumable-explicit-size
190 import os
191 import tempfile
193 import mock
194 import requests
195 import http.client
197 from google.resumable_media.requests import ResumableUpload
199 upload_url = 'http://test.invalid'
200 chunk_size = 3 * 1024 * 1024 # 3MB
201 upload = ResumableUpload(upload_url, chunk_size)
203 file_desc, filename = tempfile.mkstemp()
204 os.close(file_desc)
206 data = b'some bytes!'
207 with open(filename, 'wb') as file_obj:
208 file_obj.write(data)
210 fake_response = requests.Response()
211 fake_response.status_code = int(http.client.OK)
212 fake_response._content = b''
213 resumable_url = 'http://test.invalid?upload_id=7up'
214 fake_response.headers['location'] = resumable_url
216 post_method = mock.Mock(return_value=fake_response, spec=[])
217 transport = mock.Mock(request=post_method, spec=['request'])
219 .. doctest:: resumable-explicit-size
221 >>> import os
222 >>>
223 >>> upload.total_bytes is None
224 True
225 >>>
226 >>> stream = open(filename, 'rb')
227 >>> total_bytes = os.path.getsize(filename)
228 >>> metadata = {'name': filename}
229 >>> response = upload.initiate(
230 ... transport, stream, metadata, 'text/plain',
231 ... total_bytes=total_bytes)
232 >>> response
233 <Response [200]>
234 >>>
235 >>> upload.total_bytes == total_bytes
236 True
238 .. testcleanup:: resumable-explicit-size
240 os.remove(filename)
242 If the stream is in a "final" state (i.e. it won't have any more bytes
243 written to it), the total number of bytes can be determined implicitly
244 from the ``stream`` itself:
246 .. testsetup:: resumable-implicit-size
248 import io
250 import mock
251 import requests
252 import http.client
254 from google.resumable_media.requests import ResumableUpload
256 upload_url = 'http://test.invalid'
257 chunk_size = 3 * 1024 * 1024 # 3MB
258 upload = ResumableUpload(upload_url, chunk_size)
260 fake_response = requests.Response()
261 fake_response.status_code = int(http.client.OK)
262 fake_response._content = b''
263 resumable_url = 'http://test.invalid?upload_id=7up'
264 fake_response.headers['location'] = resumable_url
266 post_method = mock.Mock(return_value=fake_response, spec=[])
267 transport = mock.Mock(request=post_method, spec=['request'])
269 data = b'some MOAR bytes!'
270 metadata = {'name': 'some-file.jpg'}
271 content_type = 'image/jpeg'
273 .. doctest:: resumable-implicit-size
275 >>> stream = io.BytesIO(data)
276 >>> response = upload.initiate(
277 ... transport, stream, metadata, content_type)
278 >>>
279 >>> upload.total_bytes == len(data)
280 True
282 If the size of the resource is **unknown** when the upload is initiated,
283 the ``stream_final`` argument can be used. This might occur if the
284 resource is being dynamically created on the client (e.g. application
285 logs). To use this argument:
287 .. testsetup:: resumable-unknown-size
289 import io
291 import mock
292 import requests
293 import http.client
295 from google.resumable_media.requests import ResumableUpload
297 upload_url = 'http://test.invalid'
298 chunk_size = 3 * 1024 * 1024 # 3MB
299 upload = ResumableUpload(upload_url, chunk_size)
301 fake_response = requests.Response()
302 fake_response.status_code = int(http.client.OK)
303 fake_response._content = b''
304 resumable_url = 'http://test.invalid?upload_id=7up'
305 fake_response.headers['location'] = resumable_url
307 post_method = mock.Mock(return_value=fake_response, spec=[])
308 transport = mock.Mock(request=post_method, spec=['request'])
310 metadata = {'name': 'some-file.jpg'}
311 content_type = 'application/octet-stream'
313 stream = io.BytesIO(b'data')
315 .. doctest:: resumable-unknown-size
317 >>> response = upload.initiate(
318 ... transport, stream, metadata, content_type,
319 ... stream_final=False)
320 >>>
321 >>> upload.total_bytes is None
322 True
324 Args:
325 upload_url (str): The URL where the resumable upload will be initiated.
326 chunk_size (int): The size of each chunk used to upload the resource.
327 headers (Optional[Mapping[str, str]]): Extra headers that should
328 be sent with the :meth:`initiate` request, e.g. headers for
329 encrypted data. These **will not** be sent with
330 :meth:`transmit_next_chunk` or :meth:`recover` requests.
331 checksum Optional([str]): The type of checksum to compute to verify
332 the integrity of the object. After the upload is complete, the
333 server-computed checksum of the resulting object will be checked
334 and google.resumable_media.common.DataCorruption will be raised on
335 a mismatch. The corrupted file will not be deleted from the remote
336 host automatically. Supported values are "md5", "crc32c" and None.
337 The default is None.
339 Attributes:
340 upload_url (str): The URL where the content will be uploaded.
342 Raises:
343 ValueError: If ``chunk_size`` is not a multiple of
344 :data:`.UPLOAD_CHUNK_SIZE`.
345 """
347 def initiate(
348 self,
349 transport,
350 stream,
351 metadata,
352 content_type,
353 total_bytes=None,
354 stream_final=True,
355 timeout=(
356 _request_helpers._DEFAULT_CONNECT_TIMEOUT,
357 _request_helpers._DEFAULT_READ_TIMEOUT,
358 ),
359 ):
360 """Initiate a resumable upload.
362 By default, this method assumes your ``stream`` is in a "final"
363 state ready to transmit. However, ``stream_final=False`` can be used
364 to indicate that the size of the resource is not known. This can happen
365 if bytes are being dynamically fed into ``stream``, e.g. if the stream
366 is attached to application logs.
368 If ``stream_final=False`` is used, :attr:`chunk_size` bytes will be
369 read from the stream every time :meth:`transmit_next_chunk` is called.
370 If one of those reads produces strictly fewer bites than the chunk
371 size, the upload will be concluded.
373 Args:
374 transport (~requests.Session): A ``requests`` object which can
375 make authenticated requests.
376 stream (IO[bytes]): The stream (i.e. file-like object) that will
377 be uploaded. The stream **must** be at the beginning (i.e.
378 ``stream.tell() == 0``).
379 metadata (Mapping[str, str]): The resource metadata, such as an
380 ACL list.
381 content_type (str): The content type of the resource, e.g. a JPEG
382 image has content type ``image/jpeg``.
383 total_bytes (Optional[int]): The total number of bytes to be
384 uploaded. If specified, the upload size **will not** be
385 determined from the stream (even if ``stream_final=True``).
386 stream_final (Optional[bool]): Indicates if the ``stream`` is
387 "final" (i.e. no more bytes will be added to it). In this case
388 we determine the upload size from the size of the stream. If
389 ``total_bytes`` is passed, this argument will be ignored.
390 timeout (Optional[Union[float, Tuple[float, float]]]):
391 The number of seconds to wait for the server response.
392 Depending on the retry strategy, a request may be repeated
393 several times using the same timeout each time.
395 Can also be passed as a tuple (connect_timeout, read_timeout).
396 See :meth:`requests.Session.request` documentation for details.
398 Returns:
399 ~requests.Response: The HTTP response returned by ``transport``.
400 """
401 method, url, payload, headers = self._prepare_initiate_request(
402 stream,
403 metadata,
404 content_type,
405 total_bytes=total_bytes,
406 stream_final=stream_final,
407 )
409 # Wrap the request business logic in a function to be retried.
410 def retriable_request():
411 result = transport.request(
412 method, url, data=payload, headers=headers, timeout=timeout
413 )
415 self._process_initiate_response(result)
417 return result
419 return _request_helpers.wait_and_retry(
420 retriable_request, self._get_status_code, self._retry_strategy
421 )
423 def transmit_next_chunk(
424 self,
425 transport,
426 timeout=(
427 _request_helpers._DEFAULT_CONNECT_TIMEOUT,
428 _request_helpers._DEFAULT_READ_TIMEOUT,
429 ),
430 ):
431 """Transmit the next chunk of the resource to be uploaded.
433 If the current upload was initiated with ``stream_final=False``,
434 this method will dynamically determine if the upload has completed.
435 The upload will be considered complete if the stream produces
436 fewer than :attr:`chunk_size` bytes when a chunk is read from it.
438 In the case of failure, an exception is thrown that preserves the
439 failed response:
441 .. testsetup:: bad-response
443 import io
445 import mock
446 import requests
447 import http.client
449 from google import resumable_media
450 import google.resumable_media.requests.upload as upload_mod
452 transport = mock.Mock(spec=['request'])
453 fake_response = requests.Response()
454 fake_response.status_code = int(http.client.BAD_REQUEST)
455 transport.request.return_value = fake_response
457 upload_url = 'http://test.invalid'
458 upload = upload_mod.ResumableUpload(
459 upload_url, resumable_media.UPLOAD_CHUNK_SIZE)
460 # Fake that the upload has been initiate()-d
461 data = b'data is here'
462 upload._stream = io.BytesIO(data)
463 upload._total_bytes = len(data)
464 upload._resumable_url = 'http://test.invalid?upload_id=nope'
466 .. doctest:: bad-response
467 :options: +NORMALIZE_WHITESPACE
469 >>> error = None
470 >>> try:
471 ... upload.transmit_next_chunk(transport)
472 ... except resumable_media.InvalidResponse as caught_exc:
473 ... error = caught_exc
474 ...
475 >>> error
476 InvalidResponse('Request failed with status code', 400,
477 'Expected one of', <HTTPStatus.OK: 200>, <HTTPStatus.PERMANENT_REDIRECT: 308>)
478 >>> error.response
479 <Response [400]>
481 Args:
482 transport (~requests.Session): A ``requests`` object which can
483 make authenticated requests.
484 timeout (Optional[Union[float, Tuple[float, float]]]):
485 The number of seconds to wait for the server response.
486 Depending on the retry strategy, a request may be repeated
487 several times using the same timeout each time.
489 Can also be passed as a tuple (connect_timeout, read_timeout).
490 See :meth:`requests.Session.request` documentation for details.
492 Returns:
493 ~requests.Response: The HTTP response returned by ``transport``.
495 Raises:
496 ~google.resumable_media.common.InvalidResponse: If the status
497 code is not 200 or http.client.PERMANENT_REDIRECT.
498 ~google.resumable_media.common.DataCorruption: If this is the final
499 chunk, a checksum validation was requested, and the checksum
500 does not match or is not available.
501 """
502 method, url, payload, headers = self._prepare_request()
504 # Wrap the request business logic in a function to be retried.
505 def retriable_request():
506 result = transport.request(
507 method, url, data=payload, headers=headers, timeout=timeout
508 )
510 self._process_resumable_response(result, len(payload))
512 return result
514 return _request_helpers.wait_and_retry(
515 retriable_request, self._get_status_code, self._retry_strategy
516 )
518 def recover(self, transport):
519 """Recover from a failure and check the status of the current upload.
521 This will verify the progress with the server and make sure the
522 current upload is in a valid state before :meth:`transmit_next_chunk`
523 can be used again. See https://cloud.google.com/storage/docs/performing-resumable-uploads#status-check
524 for more information.
526 This method can be used when a :class:`ResumableUpload` is in an
527 :attr:`~ResumableUpload.invalid` state due to a request failure.
529 Args:
530 transport (~requests.Session): A ``requests`` object which can
531 make authenticated requests.
533 Returns:
534 ~requests.Response: The HTTP response returned by ``transport``.
535 """
536 timeout = (
537 _request_helpers._DEFAULT_CONNECT_TIMEOUT,
538 _request_helpers._DEFAULT_READ_TIMEOUT,
539 )
541 method, url, payload, headers = self._prepare_recover_request()
542 # NOTE: We assume "payload is None" but pass it along anyway.
544 # Wrap the request business logic in a function to be retried.
545 def retriable_request():
546 result = transport.request(
547 method, url, data=payload, headers=headers, timeout=timeout
548 )
550 self._process_recover_response(result)
552 return result
554 return _request_helpers.wait_and_retry(
555 retriable_request, self._get_status_code, self._retry_strategy
556 )
559class XMLMPUContainer(_request_helpers.RequestsMixin, _upload.XMLMPUContainer):
560 """Initiate and close an upload using the XML MPU API.
562 An XML MPU sends an initial request and then receives an upload ID.
563 Using the upload ID, the upload is then done in numbered parts and the
564 parts can be uploaded concurrently.
566 In order to avoid concurrency issues with this container object, the
567 uploading of individual parts is handled separately, by XMLMPUPart objects
568 spawned from this container class. The XMLMPUPart objects are not
569 necessarily in the same process as the container, so they do not update the
570 container automatically.
572 MPUs are sometimes referred to as "Multipart Uploads", which is ambiguous
573 given the JSON multipart upload, so the abbreviation "MPU" will be used
574 throughout.
576 See: https://cloud.google.com/storage/docs/multipart-uploads
578 Args:
579 upload_url (str): The URL of the object (without query parameters). The
580 initiate, PUT, and finalization requests will all use this URL, with
581 varying query parameters.
582 headers (Optional[Mapping[str, str]]): Extra headers that should
583 be sent with the :meth:`initiate` request, e.g. headers for
584 encrypted data. These headers will be propagated to individual
585 XMLMPUPart objects spawned from this container as well.
587 Attributes:
588 upload_url (str): The URL where the content will be uploaded.
589 upload_id (Optional(int)): The ID of the upload from the initialization
590 response.
591 """
593 def initiate(
594 self,
595 transport,
596 content_type,
597 timeout=(
598 _request_helpers._DEFAULT_CONNECT_TIMEOUT,
599 _request_helpers._DEFAULT_READ_TIMEOUT,
600 ),
601 ):
602 """Initiate an MPU and record the upload ID.
604 Args:
605 transport (object): An object which can make authenticated
606 requests.
607 content_type (str): The content type of the resource, e.g. a JPEG
608 image has content type ``image/jpeg``.
609 timeout (Optional[Union[float, Tuple[float, float]]]):
610 The number of seconds to wait for the server response.
611 Depending on the retry strategy, a request may be repeated
612 several times using the same timeout each time.
614 Can also be passed as a tuple (connect_timeout, read_timeout).
615 See :meth:`requests.Session.request` documentation for details.
617 Returns:
618 ~requests.Response: The HTTP response returned by ``transport``.
619 """
621 method, url, payload, headers = self._prepare_initiate_request(
622 content_type,
623 )
625 # Wrap the request business logic in a function to be retried.
626 def retriable_request():
627 result = transport.request(
628 method, url, data=payload, headers=headers, timeout=timeout
629 )
631 self._process_initiate_response(result)
633 return result
635 return _request_helpers.wait_and_retry(
636 retriable_request, self._get_status_code, self._retry_strategy
637 )
639 def finalize(
640 self,
641 transport,
642 timeout=(
643 _request_helpers._DEFAULT_CONNECT_TIMEOUT,
644 _request_helpers._DEFAULT_READ_TIMEOUT,
645 ),
646 ):
647 """Finalize an MPU request with all the parts.
649 Args:
650 transport (object): An object which can make authenticated
651 requests.
652 timeout (Optional[Union[float, Tuple[float, float]]]):
653 The number of seconds to wait for the server response.
654 Depending on the retry strategy, a request may be repeated
655 several times using the same timeout each time.
657 Can also be passed as a tuple (connect_timeout, read_timeout).
658 See :meth:`requests.Session.request` documentation for details.
660 Returns:
661 ~requests.Response: The HTTP response returned by ``transport``.
662 """
663 method, url, payload, headers = self._prepare_finalize_request()
665 # Wrap the request business logic in a function to be retried.
666 def retriable_request():
667 result = transport.request(
668 method, url, data=payload, headers=headers, timeout=timeout
669 )
671 self._process_finalize_response(result)
673 return result
675 return _request_helpers.wait_and_retry(
676 retriable_request, self._get_status_code, self._retry_strategy
677 )
679 def cancel(
680 self,
681 transport,
682 timeout=(
683 _request_helpers._DEFAULT_CONNECT_TIMEOUT,
684 _request_helpers._DEFAULT_READ_TIMEOUT,
685 ),
686 ):
687 """Cancel an MPU request and permanently delete any uploaded parts.
689 This cannot be undone.
691 Args:
692 transport (object): An object which can make authenticated
693 requests.
694 timeout (Optional[Union[float, Tuple[float, float]]]):
695 The number of seconds to wait for the server response.
696 Depending on the retry strategy, a request may be repeated
697 several times using the same timeout each time.
699 Can also be passed as a tuple (connect_timeout, read_timeout).
700 See :meth:`requests.Session.request` documentation for details.
702 Returns:
703 ~requests.Response: The HTTP response returned by ``transport``.
704 """
705 method, url, payload, headers = self._prepare_cancel_request()
707 # Wrap the request business logic in a function to be retried.
708 def retriable_request():
709 result = transport.request(
710 method, url, data=payload, headers=headers, timeout=timeout
711 )
713 self._process_cancel_response(result)
715 return result
717 return _request_helpers.wait_and_retry(
718 retriable_request, self._get_status_code, self._retry_strategy
719 )
722class XMLMPUPart(_request_helpers.RequestsMixin, _upload.XMLMPUPart):
723 def upload(
724 self,
725 transport,
726 timeout=(
727 _request_helpers._DEFAULT_CONNECT_TIMEOUT,
728 _request_helpers._DEFAULT_READ_TIMEOUT,
729 ),
730 ):
731 """Upload the part.
733 Args:
734 transport (object): An object which can make authenticated
735 requests.
736 timeout (Optional[Union[float, Tuple[float, float]]]):
737 The number of seconds to wait for the server response.
738 Depending on the retry strategy, a request may be repeated
739 several times using the same timeout each time.
741 Can also be passed as a tuple (connect_timeout, read_timeout).
742 See :meth:`requests.Session.request` documentation for details.
744 Returns:
745 ~requests.Response: The HTTP response returned by ``transport``.
746 """
747 method, url, payload, headers = self._prepare_upload_request()
749 # Wrap the request business logic in a function to be retried.
750 def retriable_request():
751 result = transport.request(
752 method, url, data=payload, headers=headers, timeout=timeout
753 )
755 self._process_upload_response(result)
757 return result
759 return _request_helpers.wait_and_retry(
760 retriable_request, self._get_status_code, self._retry_strategy
761 )