Skip to content

Commit 5fed4d2

Browse files
committed
Add 'storage.batch.Batch'
A batch proxies a connection, deferring write requests.
1 parent 8e68a20 commit 5fed4d2

2 files changed

Lines changed: 544 additions & 0 deletions

File tree

gcloud/storage/batch.py

Lines changed: 208 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,208 @@
1+
# Copyright 2014 Google Inc. All rights reserved.
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.
14+
"""Batch updates / deletes of storage buckets / blobs.
15+
16+
See: https://cloud.google.com/storage/docs/json_api/v1/how-tos/batch
17+
"""
18+
from email.encoders import encode_noop
19+
from email.generator import Generator
20+
from email.mime.application import MIMEApplication
21+
from email.mime.multipart import MIMEMultipart
22+
from email.parser import Parser
23+
import io
24+
import json
25+
import sys
26+
27+
import six
28+
29+
from gcloud._localstack import _LocalStack
30+
31+
32+
_BATCHES = _LocalStack()
33+
34+
_PROXIED_ATTRS = [
35+
'_make_request',
36+
'api_request',
37+
'build_api_url',
38+
'get_all_buckets',
39+
'get_bucket',
40+
'create_bucket',
41+
'delete_bucket',
42+
]
43+
44+
45+
class MIMEApplicationHTTP(MIMEApplication):
46+
"""MIME type for ``application/http``.
47+
48+
Constructs payload from headers and body
49+
50+
:type headers: dict
51+
:param headers: HTTP headers
52+
53+
:type body: text or None
54+
:param body: HTTP payload
55+
"""
56+
def __init__(self, method, uri, headers, body):
57+
if isinstance(body, dict):
58+
body = json.dumps(body)
59+
headers['Content-Type'] = 'application/json'
60+
headers['Content-Length'] = len(body)
61+
if body is None:
62+
body = ''
63+
lines = ['%s %s HTTP/1.1' % (method, uri)]
64+
lines.extend(['%s: %s' % (key, value)
65+
for key, value in sorted(headers.items())])
66+
lines.append('')
67+
lines.append(body)
68+
payload = '\r\n'.join(lines)
69+
if sys.version_info[0] < 3: # pragma: NO COVER Python2
70+
MIMEApplication.__init__(self, payload, 'http', encode_noop)
71+
else: # pragma: NO COVER Python3
72+
super_init = super(MIMEApplicationHTTP, self).__init__
73+
super_init(payload, 'http', encode_noop)
74+
75+
76+
class Batch(object):
77+
"""Proxy an underlying connection, batching up change operations.
78+
79+
:type connection: :class:`gcloud.storage.connection.Connection`
80+
:param connection: the connection for which the batch proxies.
81+
"""
82+
def __init__(self, connection):
83+
self._connection = connection
84+
self._http = _FauxHTTP(connection)
85+
self._requests = self._responses = ()
86+
for attr in _PROXIED_ATTRS:
87+
setattr(self, attr, getattr(connection, attr))
88+
89+
def finish(self):
90+
"""Submit a single `multipart/mixed` request w/ deferred requests.
91+
92+
:rtype: list of tuples
93+
:returns: one ``(status, reason, payload)`` tuple per deferred request.
94+
:raises: ValueError if no requests have been deferred.
95+
"""
96+
deferred = self._requests = self._http.finalize()
97+
98+
if len(deferred) == 0:
99+
raise ValueError("No deferred requests")
100+
101+
multi = MIMEMultipart()
102+
103+
for method, uri, headers, body in deferred:
104+
sub = MIMEApplicationHTTP(method, uri, headers, body)
105+
multi.attach(sub)
106+
107+
# Flatten payload
108+
if six.PY3: # pragma: NO COVER Python3
109+
buf = io.StringIO()
110+
else: # pragma: NO COVER Python2
111+
buf = io.BytesIO()
112+
generator = Generator(buf, False, 0)
113+
generator.flatten(multi)
114+
payload = buf.getvalue()
115+
116+
# Strip off redundant header text
117+
_, body = payload.split('\n\n', 1)
118+
headers = dict(multi._headers)
119+
120+
url = self._connection.build_api_url('/batch')
121+
122+
_req = self._connection._make_request
123+
response, content = _req('POST', url, data=payload, headers=headers)
124+
self._responses = list(_crack_mime_response(response, content))
125+
return self._responses
126+
127+
def __enter__(self):
128+
_BATCHES.push(self)
129+
return self
130+
131+
def __exit__(self, exc_type, exc_val, exc_tb):
132+
try:
133+
if exc_type is None:
134+
self.finish()
135+
else:
136+
self._http.reset()
137+
finally:
138+
_BATCHES.pop()
139+
140+
141+
def _crack_mime_response(response, content):
142+
"""Convert response, content -> [(status, reason, payload)].
143+
"""
144+
parser = Parser()
145+
faux = ('Content-Type: %s\nMIME-Version: 1.0\n\n%s' %
146+
(response['Content-Type'], content))
147+
148+
message = parser.parsestr(faux)
149+
150+
if not isinstance(message._payload, list):
151+
raise ValueError('Bad response: not multi-part')
152+
153+
for sub in message._payload:
154+
status_line, rest = sub._payload.split('\n', 1)
155+
_, status, reason = status_line.split(' ', 2)
156+
message = parser.parsestr(rest)
157+
payload = message._payload
158+
ctype = message['Content-Type']
159+
if ctype and ctype.startswith('application/json'):
160+
payload = json.loads(payload)
161+
yield status, reason, payload
162+
163+
164+
class NoContent(object):
165+
"""Emulate an HTTP '204 No Content' response."""
166+
status = 204
167+
168+
169+
class _FauxHTTP(object):
170+
"""Emulate ``connection.http``, but store requests.
171+
172+
Only allow up to ``_MAX_BATCH_SIZE`` requests to be bathed.
173+
"""
174+
_MAX_BATCH_SIZE = 1000
175+
176+
def __init__(self, connection):
177+
self._connection = connection
178+
self._requests = []
179+
self._orig_http, connection.http = connection.http, self
180+
181+
def request(self, method, uri, headers, body):
182+
"""Emulate / proxy underlying HTTP request.
183+
184+
- Pass 'GET' requests through.
185+
186+
- Defer others for later processing
187+
"""
188+
if method == 'GET':
189+
_req = self._orig_http.request
190+
return _req(method=method, uri=uri, headers=headers, body=body)
191+
192+
if len(self._requests) >= self._MAX_BATCH_SIZE:
193+
raise ValueError("Too many deferred requests (max %d)" %
194+
self._MAX_BATCH_SIZE)
195+
196+
self._requests.append((method, uri, headers, body))
197+
return NoContent(), ''
198+
199+
def reset(self):
200+
"""Restore the connection's ``http``.
201+
"""
202+
self._connection.http = self._orig_http
203+
204+
def finalize(self):
205+
"""Restore the connection's ``http``, and return the deferred requests.
206+
"""
207+
self.reset()
208+
return self._requests

0 commit comments

Comments
 (0)