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"""
22from google.resumable_media import _upload
23from google.resumable_media.requests import _request_helpers
26class SimpleUpload(_request_helpers.RequestsMixin, _upload.SimpleUpload):
27 """Upload a resource to a Google API.
29 A **simple** media upload sends no metadata and completes the upload
30 in a single request.
32 Args:
33 upload_url (str): The URL where the content will be uploaded.
34 headers (Optional[Mapping[str, str]]): Extra headers that should
35 be sent with the request, e.g. headers for encrypted data.
37 Attributes:
38 upload_url (str): The URL where the content will be uploaded.
39 """
41 def transmit(
42 self,
43 transport,
44 data,
45 content_type,
46 timeout=(
47 _request_helpers._DEFAULT_CONNECT_TIMEOUT,
48 _request_helpers._DEFAULT_READ_TIMEOUT,
49 ),
50 ):
51 """Transmit the resource to be uploaded.
53 Args:
54 transport (~requests.Session): A ``requests`` object which can
55 make authenticated requests.
56 data (bytes): The resource content to be uploaded.
57 content_type (str): The content type of the resource, e.g. a JPEG
58 image has content type ``image/jpeg``.
59 timeout (Optional[Union[float, Tuple[float, float]]]):
60 The number of seconds to wait for the server response.
61 Depending on the retry strategy, a request may be repeated
62 several times using the same timeout each time.
64 Can also be passed as a tuple (connect_timeout, read_timeout).
65 See :meth:`requests.Session.request` documentation for details.
67 Returns:
68 ~requests.Response: The HTTP response returned by ``transport``.
69 """
70 method, url, payload, headers = self._prepare_request(data, content_type)
72 # Wrap the request business logic in a function to be retried.
73 def retriable_request():
74 result = transport.request(
75 method, url, data=payload, headers=headers, timeout=timeout
76 )
78 self._process_response(result)
80 return result
82 return _request_helpers.wait_and_retry(
83 retriable_request, self._get_status_code, self._retry_strategy
84 )
87class MultipartUpload(_request_helpers.RequestsMixin, _upload.MultipartUpload):
88 """Upload a resource with metadata to a Google API.
90 A **multipart** upload sends both metadata and the resource in a single
91 (multipart) request.
93 Args:
94 upload_url (str): The URL where the content will be uploaded.
95 headers (Optional[Mapping[str, str]]): Extra headers that should
96 be sent with the request, e.g. headers for encrypted data.
97 checksum Optional([str]): The type of checksum to compute to verify
98 the integrity of the object. The request metadata will be amended
99 to include the computed value. Using this option will override a
100 manually-set checksum value. Supported values are "md5",
101 "crc32c" and None. The default is None.
103 Attributes:
104 upload_url (str): The URL where the content will be uploaded.
105 """
107 def transmit(
108 self,
109 transport,
110 data,
111 metadata,
112 content_type,
113 timeout=(
114 _request_helpers._DEFAULT_CONNECT_TIMEOUT,
115 _request_helpers._DEFAULT_READ_TIMEOUT,
116 ),
117 ):
118 """Transmit the resource to be uploaded.
120 Args:
121 transport (~requests.Session): A ``requests`` object which can
122 make authenticated requests.
123 data (bytes): The resource content to be uploaded.
124 metadata (Mapping[str, str]): The resource metadata, such as an
125 ACL list.
126 content_type (str): The content type of the resource, e.g. a JPEG
127 image has content type ``image/jpeg``.
128 timeout (Optional[Union[float, Tuple[float, float]]]):
129 The number of seconds to wait for the server response.
130 Depending on the retry strategy, a request may be repeated
131 several times using the same timeout each time.
133 Can also be passed as a tuple (connect_timeout, read_timeout).
134 See :meth:`requests.Session.request` documentation for details.
136 Returns:
137 ~requests.Response: The HTTP response returned by ``transport``.
138 """
139 method, url, payload, headers = self._prepare_request(
140 data, metadata, content_type
141 )
143 # Wrap the request business logic in a function to be retried.
144 def retriable_request():
145 result = transport.request(
146 method, url, data=payload, headers=headers, timeout=timeout
147 )
149 self._process_response(result)
151 return result
153 return _request_helpers.wait_and_retry(
154 retriable_request, self._get_status_code, self._retry_strategy
155 )
158class ResumableUpload(_request_helpers.RequestsMixin, _upload.ResumableUpload):
159 """Initiate and fulfill a resumable upload to a Google API.
161 A **resumable** upload sends an initial request with the resource metadata
162 and then gets assigned an upload ID / upload URL to send bytes to.
163 Using the upload URL, the upload is then done in chunks (determined by
164 the user) until all bytes have been uploaded.
166 When constructing a resumable upload, only the resumable upload URL and
167 the chunk size are required:
169 .. testsetup:: resumable-constructor
171 bucket = 'bucket-foo'
173 .. doctest:: resumable-constructor
175 >>> from google.resumable_media.requests import ResumableUpload
176 >>>
177 >>> url_template = (
178 ... 'https://www.googleapis.com/upload/storage/v1/b/{bucket}/o?'
179 ... 'uploadType=resumable')
180 >>> upload_url = url_template.format(bucket=bucket)
181 >>>
182 >>> chunk_size = 3 * 1024 * 1024 # 3MB
183 >>> upload = ResumableUpload(upload_url, chunk_size)
185 When initiating an upload (via :meth:`initiate`), the caller is expected
186 to pass the resource being uploaded as a file-like ``stream``. If the size
187 of the resource is explicitly known, it can be passed in directly:
189 .. testsetup:: resumable-explicit-size
191 import os
192 import tempfile
194 import mock
195 import requests
196 import http.client
198 from google.resumable_media.requests import ResumableUpload
200 upload_url = 'http://test.invalid'
201 chunk_size = 3 * 1024 * 1024 # 3MB
202 upload = ResumableUpload(upload_url, chunk_size)
204 file_desc, filename = tempfile.mkstemp()
205 os.close(file_desc)
207 data = b'some bytes!'
208 with open(filename, 'wb') as file_obj:
209 file_obj.write(data)
211 fake_response = requests.Response()
212 fake_response.status_code = int(http.client.OK)
213 fake_response._content = b''
214 resumable_url = 'http://test.invalid?upload_id=7up'
215 fake_response.headers['location'] = resumable_url
217 post_method = mock.Mock(return_value=fake_response, spec=[])
218 transport = mock.Mock(request=post_method, spec=['request'])
220 .. doctest:: resumable-explicit-size
222 >>> import os
223 >>>
224 >>> upload.total_bytes is None
225 True
226 >>>
227 >>> stream = open(filename, 'rb')
228 >>> total_bytes = os.path.getsize(filename)
229 >>> metadata = {'name': filename}
230 >>> response = upload.initiate(
231 ... transport, stream, metadata, 'text/plain',
232 ... total_bytes=total_bytes)
233 >>> response
234 <Response [200]>
235 >>>
236 >>> upload.total_bytes == total_bytes
237 True
239 .. testcleanup:: resumable-explicit-size
241 os.remove(filename)
243 If the stream is in a "final" state (i.e. it won't have any more bytes
244 written to it), the total number of bytes can be determined implicitly
245 from the ``stream`` itself:
247 .. testsetup:: resumable-implicit-size
249 import io
251 import mock
252 import requests
253 import http.client
255 from google.resumable_media.requests import ResumableUpload
257 upload_url = 'http://test.invalid'
258 chunk_size = 3 * 1024 * 1024 # 3MB
259 upload = ResumableUpload(upload_url, chunk_size)
261 fake_response = requests.Response()
262 fake_response.status_code = int(http.client.OK)
263 fake_response._content = b''
264 resumable_url = 'http://test.invalid?upload_id=7up'
265 fake_response.headers['location'] = resumable_url
267 post_method = mock.Mock(return_value=fake_response, spec=[])
268 transport = mock.Mock(request=post_method, spec=['request'])
270 data = b'some MOAR bytes!'
271 metadata = {'name': 'some-file.jpg'}
272 content_type = 'image/jpeg'
274 .. doctest:: resumable-implicit-size
276 >>> stream = io.BytesIO(data)
277 >>> response = upload.initiate(
278 ... transport, stream, metadata, content_type)
279 >>>
280 >>> upload.total_bytes == len(data)
281 True
283 If the size of the resource is **unknown** when the upload is initiated,
284 the ``stream_final`` argument can be used. This might occur if the
285 resource is being dynamically created on the client (e.g. application
286 logs). To use this argument:
288 .. testsetup:: resumable-unknown-size
290 import io
292 import mock
293 import requests
294 import http.client
296 from google.resumable_media.requests import ResumableUpload
298 upload_url = 'http://test.invalid'
299 chunk_size = 3 * 1024 * 1024 # 3MB
300 upload = ResumableUpload(upload_url, chunk_size)
302 fake_response = requests.Response()
303 fake_response.status_code = int(http.client.OK)
304 fake_response._content = b''
305 resumable_url = 'http://test.invalid?upload_id=7up'
306 fake_response.headers['location'] = resumable_url
308 post_method = mock.Mock(return_value=fake_response, spec=[])
309 transport = mock.Mock(request=post_method, spec=['request'])
311 metadata = {'name': 'some-file.jpg'}
312 content_type = 'application/octet-stream'
314 stream = io.BytesIO(b'data')
316 .. doctest:: resumable-unknown-size
318 >>> response = upload.initiate(
319 ... transport, stream, metadata, content_type,
320 ... stream_final=False)
321 >>>
322 >>> upload.total_bytes is None
323 True
325 Args:
326 upload_url (str): The URL where the resumable upload will be initiated.
327 chunk_size (int): The size of each chunk used to upload the resource.
328 headers (Optional[Mapping[str, str]]): Extra headers that should
329 be sent with the :meth:`initiate` request, e.g. headers for
330 encrypted data. These **will not** be sent with
331 :meth:`transmit_next_chunk` or :meth:`recover` requests.
332 checksum Optional([str]): The type of checksum to compute to verify
333 the integrity of the object. After the upload is complete, the
334 server-computed checksum of the resulting object will be checked
335 and google.resumable_media.common.DataCorruption will be raised on
336 a mismatch. The corrupted file will not be deleted from the remote
337 host automatically. Supported values are "md5", "crc32c" and None.
338 The default is None.
340 Attributes:
341 upload_url (str): The URL where the content will be uploaded.
343 Raises:
344 ValueError: If ``chunk_size`` is not a multiple of
345 :data:`.UPLOAD_CHUNK_SIZE`.
346 """
348 def initiate(
349 self,
350 transport,
351 stream,
352 metadata,
353 content_type,
354 total_bytes=None,
355 stream_final=True,
356 timeout=(
357 _request_helpers._DEFAULT_CONNECT_TIMEOUT,
358 _request_helpers._DEFAULT_READ_TIMEOUT,
359 ),
360 ):
361 """Initiate a resumable upload.
363 By default, this method assumes your ``stream`` is in a "final"
364 state ready to transmit. However, ``stream_final=False`` can be used
365 to indicate that the size of the resource is not known. This can happen
366 if bytes are being dynamically fed into ``stream``, e.g. if the stream
367 is attached to application logs.
369 If ``stream_final=False`` is used, :attr:`chunk_size` bytes will be
370 read from the stream every time :meth:`transmit_next_chunk` is called.
371 If one of those reads produces strictly fewer bites than the chunk
372 size, the upload will be concluded.
374 Args:
375 transport (~requests.Session): A ``requests`` object which can
376 make authenticated requests.
377 stream (IO[bytes]): The stream (i.e. file-like object) that will
378 be uploaded. The stream **must** be at the beginning (i.e.
379 ``stream.tell() == 0``).
380 metadata (Mapping[str, str]): The resource metadata, such as an
381 ACL list.
382 content_type (str): The content type of the resource, e.g. a JPEG
383 image has content type ``image/jpeg``.
384 total_bytes (Optional[int]): The total number of bytes to be
385 uploaded. If specified, the upload size **will not** be
386 determined from the stream (even if ``stream_final=True``).
387 stream_final (Optional[bool]): Indicates if the ``stream`` is
388 "final" (i.e. no more bytes will be added to it). In this case
389 we determine the upload size from the size of the stream. If
390 ``total_bytes`` is passed, this argument will be ignored.
391 timeout (Optional[Union[float, Tuple[float, float]]]):
392 The number of seconds to wait for the server response.
393 Depending on the retry strategy, a request may be repeated
394 several times using the same timeout each time.
396 Can also be passed as a tuple (connect_timeout, read_timeout).
397 See :meth:`requests.Session.request` documentation for details.
399 Returns:
400 ~requests.Response: The HTTP response returned by ``transport``.
401 """
402 method, url, payload, headers = self._prepare_initiate_request(
403 stream,
404 metadata,
405 content_type,
406 total_bytes=total_bytes,
407 stream_final=stream_final,
408 )
410 # Wrap the request business logic in a function to be retried.
411 def retriable_request():
412 result = transport.request(
413 method, url, data=payload, headers=headers, timeout=timeout
414 )
416 self._process_initiate_response(result)
418 return result
420 return _request_helpers.wait_and_retry(
421 retriable_request, self._get_status_code, self._retry_strategy
422 )
424 def transmit_next_chunk(
425 self,
426 transport,
427 timeout=(
428 _request_helpers._DEFAULT_CONNECT_TIMEOUT,
429 _request_helpers._DEFAULT_READ_TIMEOUT,
430 ),
431 ):
432 """Transmit the next chunk of the resource to be uploaded.
434 If the current upload was initiated with ``stream_final=False``,
435 this method will dynamically determine if the upload has completed.
436 The upload will be considered complete if the stream produces
437 fewer than :attr:`chunk_size` bytes when a chunk is read from it.
439 In the case of failure, an exception is thrown that preserves the
440 failed response:
442 .. testsetup:: bad-response
444 import io
446 import mock
447 import requests
448 import http.client
450 from google import resumable_media
451 import google.resumable_media.requests.upload as upload_mod
453 transport = mock.Mock(spec=['request'])
454 fake_response = requests.Response()
455 fake_response.status_code = int(http.client.BAD_REQUEST)
456 transport.request.return_value = fake_response
458 upload_url = 'http://test.invalid'
459 upload = upload_mod.ResumableUpload(
460 upload_url, resumable_media.UPLOAD_CHUNK_SIZE)
461 # Fake that the upload has been initiate()-d
462 data = b'data is here'
463 upload._stream = io.BytesIO(data)
464 upload._total_bytes = len(data)
465 upload._resumable_url = 'http://test.invalid?upload_id=nope'
467 .. doctest:: bad-response
468 :options: +NORMALIZE_WHITESPACE
470 >>> error = None
471 >>> try:
472 ... upload.transmit_next_chunk(transport)
473 ... except resumable_media.InvalidResponse as caught_exc:
474 ... error = caught_exc
475 ...
476 >>> error
477 InvalidResponse('Request failed with status code', 400,
478 'Expected one of', <HTTPStatus.OK: 200>, <HTTPStatus.PERMANENT_REDIRECT: 308>)
479 >>> error.response
480 <Response [400]>
482 Args:
483 transport (~requests.Session): A ``requests`` object which can
484 make authenticated requests.
485 timeout (Optional[Union[float, Tuple[float, float]]]):
486 The number of seconds to wait for the server response.
487 Depending on the retry strategy, a request may be repeated
488 several times using the same timeout each time.
490 Can also be passed as a tuple (connect_timeout, read_timeout).
491 See :meth:`requests.Session.request` documentation for details.
493 Returns:
494 ~requests.Response: The HTTP response returned by ``transport``.
496 Raises:
497 ~google.resumable_media.common.InvalidResponse: If the status
498 code is not 200 or http.client.PERMANENT_REDIRECT.
499 ~google.resumable_media.common.DataCorruption: If this is the final
500 chunk, a checksum validation was requested, and the checksum
501 does not match or is not available.
502 """
503 method, url, payload, headers = self._prepare_request()
505 # Wrap the request business logic in a function to be retried.
506 def retriable_request():
507 result = transport.request(
508 method, url, data=payload, headers=headers, timeout=timeout
509 )
511 self._process_resumable_response(result, len(payload))
513 return result
515 return _request_helpers.wait_and_retry(
516 retriable_request, self._get_status_code, self._retry_strategy
517 )
519 def recover(self, transport):
520 """Recover from a failure and check the status of the current upload.
522 This will verify the progress with the server and make sure the
523 current upload is in a valid state before :meth:`transmit_next_chunk`
524 can be used again. See https://cloud.google.com/storage/docs/performing-resumable-uploads#status-check
525 for more information.
527 This method can be used when a :class:`ResumableUpload` is in an
528 :attr:`~ResumableUpload.invalid` state due to a request failure.
530 Args:
531 transport (~requests.Session): A ``requests`` object which can
532 make authenticated requests.
534 Returns:
535 ~requests.Response: The HTTP response returned by ``transport``.
536 """
537 timeout = (
538 _request_helpers._DEFAULT_CONNECT_TIMEOUT,
539 _request_helpers._DEFAULT_READ_TIMEOUT,
540 )
542 method, url, payload, headers = self._prepare_recover_request()
543 # NOTE: We assume "payload is None" but pass it along anyway.
545 # Wrap the request business logic in a function to be retried.
546 def retriable_request():
547 result = transport.request(
548 method, url, data=payload, headers=headers, timeout=timeout
549 )
551 self._process_recover_response(result)
553 return result
555 return _request_helpers.wait_and_retry(
556 retriable_request, self._get_status_code, self._retry_strategy
557 )
560class XMLMPUContainer(_request_helpers.RequestsMixin, _upload.XMLMPUContainer):
561 """Initiate and close an upload using the XML MPU API.
563 An XML MPU sends an initial request and then receives an upload ID.
564 Using the upload ID, the upload is then done in numbered parts and the
565 parts can be uploaded concurrently.
567 In order to avoid concurrency issues with this container object, the
568 uploading of individual parts is handled separately, by XMLMPUPart objects
569 spawned from this container class. The XMLMPUPart objects are not
570 necessarily in the same process as the container, so they do not update the
571 container automatically.
573 MPUs are sometimes referred to as "Multipart Uploads", which is ambiguous
574 given the JSON multipart upload, so the abbreviation "MPU" will be used
575 throughout.
577 See: https://cloud.google.com/storage/docs/multipart-uploads
579 Args:
580 upload_url (str): The URL of the object (without query parameters). The
581 initiate, PUT, and finalization requests will all use this URL, with
582 varying query parameters.
583 headers (Optional[Mapping[str, str]]): Extra headers that should
584 be sent with the :meth:`initiate` request, e.g. headers for
585 encrypted data. These headers will be propagated to individual
586 XMLMPUPart objects spawned from this container as well.
588 Attributes:
589 upload_url (str): The URL where the content will be uploaded.
590 upload_id (Optional(int)): The ID of the upload from the initialization
591 response.
592 """
594 def initiate(
595 self,
596 transport,
597 content_type,
598 timeout=(
599 _request_helpers._DEFAULT_CONNECT_TIMEOUT,
600 _request_helpers._DEFAULT_READ_TIMEOUT,
601 ),
602 ):
603 """Initiate an MPU and record the upload ID.
605 Args:
606 transport (object): An object which can make authenticated
607 requests.
608 content_type (str): The content type of the resource, e.g. a JPEG
609 image has content type ``image/jpeg``.
610 timeout (Optional[Union[float, Tuple[float, float]]]):
611 The number of seconds to wait for the server response.
612 Depending on the retry strategy, a request may be repeated
613 several times using the same timeout each time.
615 Can also be passed as a tuple (connect_timeout, read_timeout).
616 See :meth:`requests.Session.request` documentation for details.
618 Returns:
619 ~requests.Response: The HTTP response returned by ``transport``.
620 """
622 method, url, payload, headers = self._prepare_initiate_request(
623 content_type,
624 )
626 # Wrap the request business logic in a function to be retried.
627 def retriable_request():
628 result = transport.request(
629 method, url, data=payload, headers=headers, timeout=timeout
630 )
632 self._process_initiate_response(result)
634 return result
636 return _request_helpers.wait_and_retry(
637 retriable_request, self._get_status_code, self._retry_strategy
638 )
640 def finalize(
641 self,
642 transport,
643 timeout=(
644 _request_helpers._DEFAULT_CONNECT_TIMEOUT,
645 _request_helpers._DEFAULT_READ_TIMEOUT,
646 ),
647 ):
648 """Finalize an MPU request with all the parts.
650 Args:
651 transport (object): An object which can make authenticated
652 requests.
653 timeout (Optional[Union[float, Tuple[float, float]]]):
654 The number of seconds to wait for the server response.
655 Depending on the retry strategy, a request may be repeated
656 several times using the same timeout each time.
658 Can also be passed as a tuple (connect_timeout, read_timeout).
659 See :meth:`requests.Session.request` documentation for details.
661 Returns:
662 ~requests.Response: The HTTP response returned by ``transport``.
663 """
664 method, url, payload, headers = self._prepare_finalize_request()
666 # Wrap the request business logic in a function to be retried.
667 def retriable_request():
668 result = transport.request(
669 method, url, data=payload, headers=headers, timeout=timeout
670 )
672 self._process_finalize_response(result)
674 return result
676 return _request_helpers.wait_and_retry(
677 retriable_request, self._get_status_code, self._retry_strategy
678 )
680 def cancel(
681 self,
682 transport,
683 timeout=(
684 _request_helpers._DEFAULT_CONNECT_TIMEOUT,
685 _request_helpers._DEFAULT_READ_TIMEOUT,
686 ),
687 ):
688 """Cancel an MPU request and permanently delete any uploaded parts.
690 This cannot be undone.
692 Args:
693 transport (object): An object which can make authenticated
694 requests.
695 timeout (Optional[Union[float, Tuple[float, float]]]):
696 The number of seconds to wait for the server response.
697 Depending on the retry strategy, a request may be repeated
698 several times using the same timeout each time.
700 Can also be passed as a tuple (connect_timeout, read_timeout).
701 See :meth:`requests.Session.request` documentation for details.
703 Returns:
704 ~requests.Response: The HTTP response returned by ``transport``.
705 """
706 method, url, payload, headers = self._prepare_cancel_request()
708 # Wrap the request business logic in a function to be retried.
709 def retriable_request():
710 result = transport.request(
711 method, url, data=payload, headers=headers, timeout=timeout
712 )
714 self._process_cancel_response(result)
716 return result
718 return _request_helpers.wait_and_retry(
719 retriable_request, self._get_status_code, self._retry_strategy
720 )
723class XMLMPUPart(_request_helpers.RequestsMixin, _upload.XMLMPUPart):
724 def upload(
725 self,
726 transport,
727 timeout=(
728 _request_helpers._DEFAULT_CONNECT_TIMEOUT,
729 _request_helpers._DEFAULT_READ_TIMEOUT,
730 ),
731 ):
732 """Upload the part.
734 Args:
735 transport (object): An object which can make authenticated
736 requests.
737 timeout (Optional[Union[float, Tuple[float, float]]]):
738 The number of seconds to wait for the server response.
739 Depending on the retry strategy, a request may be repeated
740 several times using the same timeout each time.
742 Can also be passed as a tuple (connect_timeout, read_timeout).
743 See :meth:`requests.Session.request` documentation for details.
745 Returns:
746 ~requests.Response: The HTTP response returned by ``transport``.
747 """
748 method, url, payload, headers = self._prepare_upload_request()
750 # Wrap the request business logic in a function to be retried.
751 def retriable_request():
752 result = transport.request(
753 method, url, data=payload, headers=headers, timeout=timeout
754 )
756 self._process_upload_response(result)
758 return result
760 return _request_helpers.wait_and_retry(
761 retriable_request, self._get_status_code, self._retry_strategy
762 )