requests utilities for Google Media Downloads and Resumable Uploads.
This sub-package assumes callers will use the requests library
as transport and google-auth for sending authenticated HTTP traffic
with requests.
.. _requests: http://docs.python-requests.org/ .. _google-auth: https://google-auth.readthedocs.io/
====================
Authorized Transport
To use google-auth and requests to create an authorized transport
that has read-only access to Google Cloud Storage (GCS):
.. testsetup:: get-credentials
import google.auth import google.auth.credentials as creds_mod import mock
def mock_default(scopes=None): credentials = mock.Mock(spec=creds_mod.Credentials) return credentials, 'mock-project'
Patch the default function on the module.
original_default = google.auth.default google.auth.default = mock_default
.. doctest:: get-credentials
import google.auth import google.auth.transport.requests as tr_requests
ro_scope = 'https://www.googleapis.com/auth/devstorage.read_only' credentials, _ = google.auth.default(scopes=(ro_scope,)) transport = tr_requests.AuthorizedSession(credentials) transport <google.auth.transport.requests.AuthorizedSession object at 0x...>
.. testcleanup:: get-credentials
Put back the correct default function on the module.
google.auth.default = original_default
================
Simple Downloads
To download an object from Google Cloud Storage, construct the media URL for the GCS object and download it with an authorized transport that has access to the resource:
.. testsetup:: basic-download
import mock import requests import http.client
bucket = 'bucket-foo' blob_name = 'file.txt'
fake_response = requests.Response() fake_response.status_code = int(http.client.OK) fake_response.headers['Content-Length'] = '1364156' fake_content = mock.MagicMock(spec=['len']) fake_content.len.return_value = 1364156 fake_response._content = fake_content
get_method = mock.Mock(return_value=fake_response, spec=[]) transport = mock.Mock(request=get_method, spec=['request'])
.. doctest:: basic-download
from google.resumable_media.requests import Download
url_template = ( ... 'https://www.googleapis.com/download/storage/v1/b/' ... '{bucket}/o/{blob_name}?alt=media') media_url = url_template.format( ... bucket=bucket, blob_name=blob_name)
download = Download(media_url) response = download.consume(transport) download.finished True response <Response [200]> response.headers['Content-Length'] '1364156' len(response.content) 1364156
To download only a portion of the bytes in the object,
specify start and end byte positions (both optional):
.. testsetup:: basic-download-with-slice
import mock import requests import http.client
from google.resumable_media.requests import Download
media_url = 'http://test.invalid' start = 4096 end = 8191 slice_size = end - start + 1
fake_response = requests.Response() fake_response.status_code = int(http.client.PARTIAL_CONTENT) fake_response.headers['Content-Length'] = '{:d}'.format(slice_size) content_range = 'bytes {:d}-{:d}/1364156'.format(start, end) fake_response.headers['Content-Range'] = content_range fake_content = mock.MagicMock(spec=['len']) fake_content.len.return_value = slice_size fake_response._content = fake_content
get_method = mock.Mock(return_value=fake_response, spec=[]) transport = mock.Mock(request=get_method, spec=['request'])
.. doctest:: basic-download-with-slice
download = Download(media_url, start=4096, end=8191) response = download.consume(transport) download.finished True response <Response [206]> response.headers['Content-Length'] '4096' response.headers['Content-Range'] 'bytes 4096-8191/1364156' len(response.content) 4096
=================
Chunked Downloads
For very large objects or objects of unknown size, it may make more sense to download the object in chunks rather than all at once. This can be done to avoid dropped connections with a poor internet connection or can allow multiple chunks to be downloaded in parallel to speed up the total download.
A <xref:.ChunkedDownload> uses the same media URL and authorized
transport that a basic <xref:.Download> would use, but also
requires a chunk size and a write-able byte stream. The chunk size is used
to determine how much of the resouce to consume with each request and the
stream is to allow the resource to be written out (e.g. to disk) without
having to fit in memory all at once.
.. testsetup:: chunked-download
import io
import mock import requests import http.client
media_url = 'http://test.invalid'
fifty_mb = 50 * 1024 * 1024 one_gb = 1024 * 1024 * 1024 fake_response = requests.Response() fake_response.status_code = int(http.client.PARTIAL_CONTENT) fake_response.headers['Content-Length'] = '{:d}'.format(fifty_mb) content_range = 'bytes 0-{:d}/{:d}'.format(fifty_mb - 1, one_gb) fake_response.headers['Content-Range'] = content_range fake_content_begin = b'The beginning of the chunk...' fake_content = fake_content_begin + b'1' * (fifty_mb - 29) fake_response._content = fake_content
get_method = mock.Mock(return_value=fake_response, spec=[]) transport = mock.Mock(request=get_method, spec=['request'])
.. doctest:: chunked-download
from google.resumable_media.requests import ChunkedDownload
chunk_size = 50 * 1024 * 1024 # 50MB stream = io.BytesIO() download = ChunkedDownload( ... media_url, chunk_size, stream)
Check the state of the download before starting.
download.bytes_downloaded 0 download.total_bytes is None True response = download.consume_next_chunk(transport)
Check the state of the download after consuming one chunk.
download.finished False download.bytes_downloaded # chunk_size 52428800 download.total_bytes # 1GB 1073741824 response <Response [206]> response.headers['Content-Length'] '52428800' response.headers['Content-Range'] 'bytes 0-52428799/1073741824' len(response.content) == chunk_size True stream.seek(0) 0 stream.read(29) b'The beginning of the chunk...'
The download will change it's finished status to :data:True
once the final chunk is consumed. In some cases, the final chunk may
not be the same size as the other chunks:
.. testsetup:: chunked-download-end
import mock import requests import http.client
from google.resumable_media.requests import ChunkedDownload
media_url = 'http://test.invalid'
fifty_mb = 50 * 1024 * 1024 one_gb = 1024 * 1024 * 1024 stream = mock.Mock(spec=['write']) download = ChunkedDownload(media_url, fifty_mb, stream) download._bytes_downloaded = 20 * fifty_mb download._total_bytes = one_gb
fake_response = requests.Response() fake_response.status_code = int(http.client.PARTIAL_CONTENT) slice_size = one_gb - 20 * fifty_mb fake_response.headers['Content-Length'] = '{:d}'.format(slice_size) content_range = 'bytes {:d}-{:d}/{:d}'.format( 20 * fifty_mb, one_gb - 1, one_gb) fake_response.headers['Content-Range'] = content_range fake_content = mock.MagicMock(spec=['len']) fake_content.len.return_value = slice_size fake_response._content = fake_content
get_method = mock.Mock(return_value=fake_response, spec=[]) transport = mock.Mock(request=get_method, spec=['request'])
.. doctest:: chunked-download-end
The state of the download in progress.
download.finished False download.bytes_downloaded # 20 chunks at 50MB 1048576000 download.total_bytes # 1GB 1073741824 response = download.consume_next_chunk(transport)
The state of the download after consuming the final chunk.
download.finished True download.bytes_downloaded == download.total_bytes True response <Response [206]> response.headers['Content-Length'] '25165824' response.headers['Content-Range'] 'bytes 1048576000-1073741823/1073741824' len(response.content) < download.chunk_size True
In addition, a <xref:.ChunkedDownload> can also take optional
start and end byte positions.
==============
Simple Uploads
Among the three supported upload classes, the simplest is <xref:.SimpleUpload>. A simple upload should be used when the resource being uploaded is small and when there is no metadata (other than the name) associated with the resource.
.. testsetup:: simple-upload
import json
import mock import requests import http.client
bucket = 'some-bucket' blob_name = 'file.txt'
fake_response = requests.Response() fake_response.status_code = int(http.client.OK) payload = { 'bucket': bucket, 'contentType': 'text/plain', 'md5Hash': 'M0XLEsX9/sMdiI+4pB4CAQ==', 'name': blob_name, 'size': '27', } fake_response._content = json.dumps(payload).encode('utf-8')
post_method = mock.Mock(return_value=fake_response, spec=[]) transport = mock.Mock(request=post_method, spec=['request'])
.. doctest:: simple-upload :options: +NORMALIZE_WHITESPACE
from google.resumable_media.requests import SimpleUpload
url_template = ( ... 'https://www.googleapis.com/upload/storage/v1/b/{bucket}/o?' ... 'uploadType=media&' ... 'name={blob_name}') upload_url = url_template.format( ... bucket=bucket, blob_name=blob_name)
upload = SimpleUpload(upload_url) data = b'Some not too large content.' content_type = 'text/plain' response = upload.transmit(transport, data, content_type) upload.finished True response <Response [200]> json_response = response.json() json_response['bucket'] == bucket True json_response['name'] == blob_name True json_response['contentType'] == content_type True json_response['md5Hash'] 'M0XLEsX9/sMdiI+4pB4CAQ==' int(json_response['size']) == len(data) True
In the rare case that an upload fails, an :exc:.InvalidResponse
will be raised:
.. testsetup:: simple-upload-fail
import time
import mock import requests import http.client
from google import resumable_media from google.resumable_media import _helpers from google.resumable_media.requests import SimpleUpload as constructor
upload_url = 'http://test.invalid' data = b'Some not too large content.' content_type = 'text/plain'
fake_response = requests.Response() fake_response.status_code = int(http.client.SERVICE_UNAVAILABLE)
post_method = mock.Mock(return_value=fake_response, spec=[]) transport = mock.Mock(request=post_method, spec=['request'])
time_sleep = time.sleep def dont_sleep(seconds): raise RuntimeError('No sleep', seconds)
def SimpleUpload(args, **kwargs): upload = constructor(args, **kwargs)
Mock the cumulative sleep to avoid retries (and time.sleep()).
upload._retry_strategy = resumable_media.RetryStrategy(
max_cumulative_retry=-1.0)
return upload
time.sleep = dont_sleep
.. doctest:: simple-upload-fail :options: +NORMALIZE_WHITESPACE
upload = SimpleUpload(upload_url) error = None try: ... upload.transmit(transport, data, content_type) ... except resumable_media.InvalidResponse as caught_exc: ... error = caught_exc ... error InvalidResponse('Request failed with status code', 503, 'Expected one of', <HTTPStatus.OK: 200>) error.response <Response [503]>
upload.finished True
.. testcleanup:: simple-upload-fail
Put back the correct sleep function on the time module.
time.sleep = time_sleep
Even in the case of failure, we see that the upload is
:attr:~.SimpleUpload.finished, i.e. it cannot be re-used.
=================
Multipart Uploads
After the simple upload, the <xref:.MultipartUpload> can be used to achieve essentially the same task. However, a multipart upload allows some metadata about the resource to be sent along as well. (This is the "multi": we send a first part with the metadata and a second part with the actual bytes in the resource.)
Usage is similar to the simple upload, but <xref:.MultipartUpload.transmit>
accepts an extra required argument: metadata.
.. testsetup:: multipart-upload
import json
import mock import requests import http.client
bucket = 'some-bucket' blob_name = 'file.txt' data = b'Some not too large content.' content_type = 'text/plain'
fake_response = requests.Response() fake_response.status_code = int(http.client.OK) payload = { 'bucket': bucket, 'name': blob_name, 'metadata': {'color': 'grurple'}, } fake_response._content = json.dumps(payload).encode('utf-8')
post_method = mock.Mock(return_value=fake_response, spec=[]) transport = mock.Mock(request=post_method, spec=['request'])
.. doctest:: multipart-upload
from google.resumable_media.requests import MultipartUpload
url_template = ( ... 'https://www.googleapis.com/upload/storage/v1/b/{bucket}/o?' ... 'uploadType=multipart') upload_url = url_template.format(bucket=bucket)
upload = MultipartUpload(upload_url) metadata = { ... 'name': blob_name, ... 'metadata': { ... 'color': 'grurple', ... }, ... } response = upload.transmit(transport, data, metadata, content_type) upload.finished True response <Response [200]> json_response = response.json() json_response['bucket'] == bucket True json_response['name'] == blob_name True json_response['metadata'] == metadata['metadata'] True
As with the simple upload, in the case of failure an :exc:.InvalidResponse
is raised, enclosing the :attr:~.InvalidResponse.response that caused
the failure and the upload object cannot be re-used after a failure.
=================
Resumable Uploads
A <xref:.ResumableUpload> deviates from the other two upload classes: it transmits a resource over the course of multiple requests. This is intended to be used in cases where:
- the size of the resource is not known (i.e. it is generated on the fly)
- requests must be short-lived
- the client has request size limitations
- the resource is too large to fit into memory
In general, a resource should be sent in a single request to avoid
latency and reduce QPS. See GCS best practices_ for more things to
consider when using a resumable upload.
.. _GCS best practices: https://cloud.google.com/storage/docs/ best-practices#uploading
After creating a <xref:.ResumableUpload> instance, a
resumable upload session must be initiated to let the server know that
a series of chunked upload requests will be coming and to obtain an
upload_id for the session. In contrast to the other two upload classes,
<xref:.ResumableUpload.initiate> takes a byte stream as input rather
than raw bytes as data. This can be a file object, a <xref:io.BytesIO>
object or any other stream implementing the same interface.
.. testsetup:: resumable-initiate
import io
import mock import requests import http.client
bucket = 'some-bucket' blob_name = 'file.txt' data = b'Some resumable bytes.' content_type = 'text/plain'
fake_response = requests.Response() fake_response.status_code = int(http.client.OK) fake_response._content = b'' upload_id = 'ABCdef189XY_super_serious' resumable_url_template = ( 'https://www.googleapis.com/upload/storage/v1/b/{bucket}' '/o?uploadType=resumable&upload_id={upload_id}') resumable_url = resumable_url_template.format( bucket=bucket, upload_id=upload_id) fake_response.headers['location'] = resumable_url fake_response.headers['x-guploader-uploadid'] = upload_id
post_method = mock.Mock(return_value=fake_response, spec=[]) transport = mock.Mock(request=post_method, spec=['request'])
.. doctest:: resumable-initiate
from google.resumable_media.requests import ResumableUpload
url_template = ( ... 'https://www.googleapis.com/upload/storage/v1/b/{bucket}/o?' ... 'uploadType=resumable') upload_url = url_template.format(bucket=bucket)
chunk_size = 1024 * 1024 # 1MB upload = ResumableUpload(upload_url, chunk_size) stream = io.BytesIO(data)
The upload doesn't know how "big" it is until seeing a stream.
upload.total_bytes is None True metadata = {'name': blob_name} response = upload.initiate(transport, stream, metadata, content_type) response <Response [200]> upload.resumable_url == response.headers['Location'] True upload.total_bytes == len(data) True upload_id = response.headers['X-GUploader-UploadID'] upload_id 'ABCdef189XY_super_serious' upload.resumable_url == upload_url + '&upload_id=' + upload_id True
Once a <xref:.ResumableUpload> has been initiated, the resource is transmitted in chunks until completion:
.. testsetup:: resumable-transmit
import io import json
import mock import requests import http.client
from google import resumable_media import google.resumable_media.requests.upload as upload_mod
data = b'01234567891' stream = io.BytesIO(data)
Create an "already initiated" upload.
upload_url = 'http://test.invalid' chunk_size = 256 * 1024 # 256KB upload = upload_mod.ResumableUpload(upload_url, chunk_size) upload._resumable_url = 'http://test.invalid?upload_id=mocked' upload._stream = stream upload._content_type = 'text/plain' upload._total_bytes = len(data)
After-the-fact update the chunk size so that len(data)
is split into three.
upload._chunk_size = 4
Make three fake responses.
fake_response0 = requests.Response() fake_response0.status_code = http.client.PERMANENT_REDIRECT fake_response0.headers['range'] = 'bytes=0-3'
fake_response1 = requests.Response() fake_response1.status_code = http.client.PERMANENT_REDIRECT fake_response1.headers['range'] = 'bytes=0-7'
fake_response2 = requests.Response() fake_response2.status_code = int(http.client.OK) bucket = 'some-bucket' blob_name = 'file.txt' payload = { 'bucket': bucket, 'name': blob_name, 'size': '{:d}'.format(len(data)), } fake_response2._content = json.dumps(payload).encode('utf-8')
Use the fake responses to mock a transport.
responses = [fake_response0, fake_response1, fake_response2] put_method = mock.Mock(side_effect=responses, spec=[]) transport = mock.Mock(request=put_method, spec=['request'])
.. doctest:: resumable-transmit
response0 = upload.transmit_next_chunk(transport) response0 <Response [308]> upload.finished False upload.bytes_uploaded == upload.chunk_size True
response1 = upload.transmit_next_chunk(transport) response1 <Response [308]> upload.finished False upload.bytes_uploaded == 2 * upload.chunk_size True
response2 = upload.transmit_next_chunk(transport) response2 <Response [200]> upload.finished True upload.bytes_uploaded == upload.total_bytes True json_response = response2.json() json_response['bucket'] == bucket True json_response['name'] == blob_name True