Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/google/cloud/storage/_media/_upload.py: 31%

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. 

14 

15"""Virtual bases classes for uploading media via Google APIs. 

16 

17Supported here are: 

18 

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""" 

23 

24import http.client 

25import json 

26import os 

27import random 

28import re 

29import sys 

30import urllib.parse 

31 

32from google.cloud.storage._media import _helpers 

33from google.cloud.storage._media import UPLOAD_CHUNK_SIZE 

34from google.cloud.storage.exceptions import InvalidResponse 

35from google.cloud.storage.exceptions import DataCorruption 

36from google.cloud.storage.retry import DEFAULT_RETRY 

37 

38from xml.etree import ElementTree 

39 

40 

41_CONTENT_TYPE_HEADER = "content-type" 

42_CONTENT_RANGE_TEMPLATE = "bytes {:d}-{:d}/{:d}" 

43_RANGE_UNKNOWN_TEMPLATE = "bytes {:d}-{:d}/*" 

44_EMPTY_RANGE_TEMPLATE = "bytes */{:d}" 

45_BOUNDARY_WIDTH = len(str(sys.maxsize - 1)) 

46_BOUNDARY_FORMAT = "==============={{:0{:d}d}}==".format(_BOUNDARY_WIDTH) 

47_MULTIPART_SEP = b"--" 

48_CRLF = b"\r\n" 

49_MULTIPART_BEGIN = b"\r\ncontent-type: application/json; charset=UTF-8\r\n\r\n" 

50_RELATED_HEADER = b'multipart/related; boundary="' 

51_BYTES_RANGE_RE = re.compile(r"bytes=0-(?P<end_byte>\d+)", flags=re.IGNORECASE) 

52_STREAM_ERROR_TEMPLATE = ( 

53 "Bytes stream is in unexpected state. " 

54 "The local stream has had {:d} bytes read from it while " 

55 "{:d} bytes have already been updated (they should match)." 

56) 

57_STREAM_READ_PAST_TEMPLATE = ( 

58 "{:d} bytes have been read from the stream, which exceeds " 

59 "the expected total {:d}." 

60) 

61_DELETE = "DELETE" 

62_POST = "POST" 

63_PUT = "PUT" 

64_UPLOAD_CHECKSUM_MISMATCH_MESSAGE = ( 

65 "The computed ``{}`` checksum, ``{}``, and the checksum reported by the " 

66 "remote host, ``{}``, did not match." 

67) 

68_UPLOAD_METADATA_NO_APPROPRIATE_CHECKSUM_MESSAGE = ( 

69 "Response metadata had no ``{}`` value; checksum could not be validated." 

70) 

71_UPLOAD_HEADER_NO_APPROPRIATE_CHECKSUM_MESSAGE = ( 

72 "Response headers had no ``{}`` value; checksum could not be validated." 

73) 

74_MPU_INITIATE_QUERY = "?uploads" 

75_MPU_PART_QUERY_TEMPLATE = "?partNumber={part}&uploadId={upload_id}" 

76_S3_COMPAT_XML_NAMESPACE = "{http://s3.amazonaws.com/doc/2006-03-01/}" 

77_UPLOAD_ID_NODE = "UploadId" 

78_MPU_FINAL_QUERY_TEMPLATE = "?uploadId={upload_id}" 

79 

80 

81class UploadBase(object): 

82 """Base class for upload helpers. 

83 

84 Defines core shared behavior across different upload types. 

85 

86 Args: 

87 upload_url (str): The URL where the content will be uploaded. 

88 headers (Optional[Mapping[str, str]]): Extra headers that should 

89 be sent with the request, e.g. headers for encrypted data. 

90 retry (Optional[google.api_core.retry.Retry]): How to retry the 

91 RPC. A None value will disable retries. A 

92 google.api_core.retry.Retry value will enable retries, and the 

93 object will configure backoff and timeout options. 

94 

95 See the retry.py source code and docstrings in this package 

96 (google.cloud.storage.retry) for information on retry types and how 

97 to configure them. 

98 

99 Attributes: 

100 upload_url (str): The URL where the content will be uploaded. 

101 """ 

102 

103 def __init__(self, upload_url, headers=None, retry=DEFAULT_RETRY): 

104 self.upload_url = upload_url 

105 if headers is None: 

106 headers = {} 

107 self._headers = headers 

108 self._finished = False 

109 self._retry_strategy = retry 

110 

111 @property 

112 def finished(self): 

113 """bool: Flag indicating if the upload has completed.""" 

114 return self._finished 

115 

116 def _process_response(self, response): 

117 """Process the response from an HTTP request. 

118 

119 This is everything that must be done after a request that doesn't 

120 require network I/O (or other I/O). This is based on the `sans-I/O`_ 

121 philosophy. 

122 

123 Args: 

124 response (object): The HTTP response object. 

125 

126 Raises: 

127 ~google.cloud.storage.exceptions.InvalidResponse: If the status 

128 code is not 200. 

129 

130 .. _sans-I/O: https://sans-io.readthedocs.io/ 

131 """ 

132 # Tombstone the current upload so it cannot be used again (in either 

133 # failure or success). 

134 self._finished = True 

135 _helpers.require_status_code(response, (http.client.OK,), self._get_status_code) 

136 

137 @staticmethod 

138 def _get_status_code(response): 

139 """Access the status code from an HTTP response. 

140 

141 Args: 

142 response (object): The HTTP response object. 

143 

144 Raises: 

145 NotImplementedError: Always, since virtual. 

146 """ 

147 raise NotImplementedError("This implementation is virtual.") 

148 

149 @staticmethod 

150 def _get_headers(response): 

151 """Access the headers from an HTTP response. 

152 

153 Args: 

154 response (object): The HTTP response object. 

155 

156 Raises: 

157 NotImplementedError: Always, since virtual. 

158 """ 

159 raise NotImplementedError("This implementation is virtual.") 

160 

161 @staticmethod 

162 def _get_body(response): 

163 """Access the response body from an HTTP response. 

164 

165 Args: 

166 response (object): The HTTP response object. 

167 

168 Raises: 

169 NotImplementedError: Always, since virtual. 

170 """ 

171 raise NotImplementedError("This implementation is virtual.") 

172 

173 

174class SimpleUpload(UploadBase): 

175 """Upload a resource to a Google API. 

176 

177 A **simple** media upload sends no metadata and completes the upload 

178 in a single request. 

179 

180 Args: 

181 upload_url (str): The URL where the content will be uploaded. 

182 headers (Optional[Mapping[str, str]]): Extra headers that should 

183 be sent with the request, e.g. headers for encrypted data. 

184 retry (Optional[google.api_core.retry.Retry]): How to retry the 

185 RPC. A None value will disable retries. A 

186 google.api_core.retry.Retry value will enable retries, and the 

187 object will configure backoff and timeout options. 

188 

189 See the retry.py source code and docstrings in this package 

190 (google.cloud.storage.retry) for information on retry types and how 

191 to configure them. 

192 

193 Attributes: 

194 upload_url (str): The URL where the content will be uploaded. 

195 """ 

196 

197 def _prepare_request(self, data, content_type): 

198 """Prepare the contents of an HTTP request. 

199 

200 This is everything that must be done before a request that doesn't 

201 require network I/O (or other I/O). This is based on the `sans-I/O`_ 

202 philosophy. 

203 

204 .. note: 

205 

206 This method will be used only once, so ``headers`` will be 

207 mutated by having a new key added to it. 

208 

209 Args: 

210 data (bytes): The resource content to be uploaded. 

211 content_type (str): The content type for the request. 

212 

213 Returns: 

214 Tuple[str, str, bytes, Mapping[str, str]]: The quadruple 

215 

216 * HTTP verb for the request (always POST) 

217 * the URL for the request 

218 * the body of the request 

219 * headers for the request 

220 

221 Raises: 

222 ValueError: If the current upload has already finished. 

223 TypeError: If ``data`` isn't bytes. 

224 

225 .. _sans-I/O: https://sans-io.readthedocs.io/ 

226 """ 

227 if self.finished: 

228 raise ValueError("An upload can only be used once.") 

229 

230 if not isinstance(data, bytes): 

231 raise TypeError("`data` must be bytes, received", type(data)) 

232 self._headers[_CONTENT_TYPE_HEADER] = content_type 

233 return _POST, self.upload_url, data, self._headers 

234 

235 def transmit(self, transport, data, content_type, timeout=None): 

236 """Transmit the resource to be uploaded. 

237 

238 Args: 

239 transport (object): An object which can make authenticated 

240 requests. 

241 data (bytes): The resource content to be uploaded. 

242 content_type (str): The content type of the resource, e.g. a JPEG 

243 image has content type ``image/jpeg``. 

244 timeout (Optional[Union[float, Tuple[float, float]]]): 

245 The number of seconds to wait for the server response. 

246 Depending on the retry strategy, a request may be repeated 

247 several times using the same timeout each time. 

248 

249 Can also be passed as a tuple (connect_timeout, read_timeout). 

250 See :meth:`requests.Session.request` documentation for details. 

251 

252 Raises: 

253 NotImplementedError: Always, since virtual. 

254 """ 

255 raise NotImplementedError("This implementation is virtual.") 

256 

257 

258class MultipartUpload(UploadBase): 

259 """Upload a resource with metadata to a Google API. 

260 

261 A **multipart** upload sends both metadata and the resource in a single 

262 (multipart) request. 

263 

264 Args: 

265 upload_url (str): The URL where the content will be uploaded. 

266 headers (Optional[Mapping[str, str]]): Extra headers that should 

267 be sent with the request, e.g. headers for encrypted data. 

268 checksum Optional([str]): The type of checksum to compute to verify 

269 the integrity of the object. The request metadata will be amended 

270 to include the computed value. Using this option will override a 

271 manually-set checksum value. Supported values are "md5", 

272 "crc32c", "auto", and None. The default is "auto", which will try 

273 to detect if the C extension for crc32c is installed and fall back 

274 to md5 otherwise. 

275 retry (Optional[google.api_core.retry.Retry]): How to retry the 

276 RPC. A None value will disable retries. A 

277 google.api_core.retry.Retry value will enable retries, and the 

278 object will configure backoff and timeout options. 

279 

280 See the retry.py source code and docstrings in this package 

281 (google.cloud.storage.retry) for information on retry types and how 

282 to configure them. 

283 

284 Attributes: 

285 upload_url (str): The URL where the content will be uploaded. 

286 """ 

287 

288 def __init__(self, upload_url, headers=None, checksum="auto", retry=DEFAULT_RETRY): 

289 super(MultipartUpload, self).__init__(upload_url, headers=headers, retry=retry) 

290 self._checksum_type = checksum 

291 if self._checksum_type == "auto": 

292 self._checksum_type = ( 

293 "crc32c" if _helpers._is_crc32c_available_and_fast() else "md5" 

294 ) 

295 

296 def _prepare_request(self, data, metadata, content_type): 

297 """Prepare the contents of an HTTP request. 

298 

299 This is everything that must be done before a request that doesn't 

300 require network I/O (or other I/O). This is based on the `sans-I/O`_ 

301 philosophy. 

302 

303 .. note: 

304 

305 This method will be used only once, so ``headers`` will be 

306 mutated by having a new key added to it. 

307 

308 Args: 

309 data (bytes): The resource content to be uploaded. 

310 metadata (Mapping[str, str]): The resource metadata, such as an 

311 ACL list. 

312 content_type (str): The content type of the resource, e.g. a JPEG 

313 image has content type ``image/jpeg``. 

314 

315 Returns: 

316 Tuple[str, str, bytes, Mapping[str, str]]: The quadruple 

317 

318 * HTTP verb for the request (always POST) 

319 * the URL for the request 

320 * the body of the request 

321 * headers for the request 

322 

323 Raises: 

324 ValueError: If the current upload has already finished. 

325 TypeError: If ``data`` isn't bytes. 

326 

327 .. _sans-I/O: https://sans-io.readthedocs.io/ 

328 """ 

329 if self.finished: 

330 raise ValueError("An upload can only be used once.") 

331 

332 if not isinstance(data, bytes): 

333 raise TypeError("`data` must be bytes, received", type(data)) 

334 

335 checksum_object = _helpers._get_checksum_object(self._checksum_type) 

336 if checksum_object is not None: 

337 checksum_object.update(data) 

338 actual_checksum = _helpers.prepare_checksum_digest(checksum_object.digest()) 

339 metadata_key = _helpers._get_metadata_key(self._checksum_type) 

340 metadata[metadata_key] = actual_checksum 

341 

342 content, multipart_boundary = construct_multipart_request( 

343 data, metadata, content_type 

344 ) 

345 multipart_content_type = _RELATED_HEADER + multipart_boundary + b'"' 

346 self._headers[_CONTENT_TYPE_HEADER] = multipart_content_type 

347 

348 return _POST, self.upload_url, content, self._headers 

349 

350 def transmit(self, transport, data, metadata, content_type, timeout=None): 

351 """Transmit the resource to be uploaded. 

352 

353 Args: 

354 transport (object): An object which can make authenticated 

355 requests. 

356 data (bytes): The resource content to be uploaded. 

357 metadata (Mapping[str, str]): The resource metadata, such as an 

358 ACL list. 

359 content_type (str): The content type of the resource, e.g. a JPEG 

360 image has content type ``image/jpeg``. 

361 timeout (Optional[Union[float, Tuple[float, float]]]): 

362 The number of seconds to wait for the server response. 

363 Depending on the retry strategy, a request may be repeated 

364 several times using the same timeout each time. 

365 

366 Can also be passed as a tuple (connect_timeout, read_timeout). 

367 See :meth:`requests.Session.request` documentation for details. 

368 

369 Raises: 

370 NotImplementedError: Always, since virtual. 

371 """ 

372 raise NotImplementedError("This implementation is virtual.") 

373 

374 

375class ResumableUpload(UploadBase): 

376 """Initiate and fulfill a resumable upload to a Google API. 

377 

378 A **resumable** upload sends an initial request with the resource metadata 

379 and then gets assigned an upload ID / upload URL to send bytes to. 

380 Using the upload URL, the upload is then done in chunks (determined by 

381 the user) until all bytes have been uploaded. 

382 

383 Args: 

384 upload_url (str): The URL where the resumable upload will be initiated. 

385 chunk_size (int): The size of each chunk used to upload the resource. 

386 headers (Optional[Mapping[str, str]]): Extra headers that should 

387 be sent with every request. 

388 checksum Optional([str]): The type of checksum to compute to verify 

389 the integrity of the object. After the upload is complete, the 

390 server-computed checksum of the resulting object will be checked 

391 and google.cloud.storage.exceptions.DataCorruption will be raised on 

392 a mismatch. The corrupted file will not be deleted from the remote 

393 host automatically. Supported values are "md5", "crc32c", "auto", 

394 and None. The default is "auto", which will try to detect if the C 

395 extension for crc32c is installed and fall back to md5 otherwise. 

396 retry (Optional[google.api_core.retry.Retry]): How to retry the 

397 RPC. A None value will disable retries. A 

398 google.api_core.retry.Retry value will enable retries, and the 

399 object will configure backoff and timeout options. 

400 

401 See the retry.py source code and docstrings in this package 

402 (google.cloud.storage.retry) for information on retry types and how 

403 to configure them. 

404 

405 Attributes: 

406 upload_url (str): The URL where the content will be uploaded. 

407 

408 Raises: 

409 ValueError: If ``chunk_size`` is not a multiple of 

410 :data:`.UPLOAD_CHUNK_SIZE`. 

411 """ 

412 

413 def __init__( 

414 self, upload_url, chunk_size, checksum="auto", headers=None, retry=DEFAULT_RETRY 

415 ): 

416 super(ResumableUpload, self).__init__(upload_url, headers=headers, retry=retry) 

417 if chunk_size % UPLOAD_CHUNK_SIZE != 0: 

418 raise ValueError( 

419 "{} KB must divide chunk size".format(UPLOAD_CHUNK_SIZE / 1024) 

420 ) 

421 self._chunk_size = chunk_size 

422 self._stream = None 

423 self._content_type = None 

424 self._bytes_uploaded = 0 

425 self._bytes_checksummed = 0 

426 self._checksum_type = checksum 

427 if self._checksum_type == "auto": 

428 self._checksum_type = ( 

429 "crc32c" if _helpers._is_crc32c_available_and_fast() else "md5" 

430 ) 

431 self._checksum_object = None 

432 self._total_bytes = None 

433 self._resumable_url = None 

434 self._invalid = False 

435 

436 @property 

437 def invalid(self): 

438 """bool: Indicates if the upload is in an invalid state. 

439 

440 This will occur if a call to :meth:`transmit_next_chunk` fails. 

441 To recover from such a failure, call :meth:`recover`. 

442 """ 

443 return self._invalid 

444 

445 @property 

446 def chunk_size(self): 

447 """int: The size of each chunk used to upload the resource.""" 

448 return self._chunk_size 

449 

450 @property 

451 def resumable_url(self): 

452 """Optional[str]: The URL of the in-progress resumable upload.""" 

453 return self._resumable_url 

454 

455 @property 

456 def bytes_uploaded(self): 

457 """int: Number of bytes that have been uploaded.""" 

458 return self._bytes_uploaded 

459 

460 @property 

461 def total_bytes(self): 

462 """Optional[int]: The total number of bytes to be uploaded. 

463 

464 If this upload is initiated (via :meth:`initiate`) with 

465 ``stream_final=True``, this value will be populated based on the size 

466 of the ``stream`` being uploaded. (By default ``stream_final=True``.) 

467 

468 If this upload is initiated with ``stream_final=False``, 

469 :attr:`total_bytes` will be :data:`None` since it cannot be 

470 determined from the stream. 

471 """ 

472 return self._total_bytes 

473 

474 def _prepare_initiate_request( 

475 self, stream, metadata, content_type, total_bytes=None, stream_final=True 

476 ): 

477 """Prepare the contents of HTTP request to initiate upload. 

478 

479 This is everything that must be done before a request that doesn't 

480 require network I/O (or other I/O). This is based on the `sans-I/O`_ 

481 philosophy. 

482 

483 Args: 

484 stream (IO[bytes]): The stream (i.e. file-like object) that will 

485 be uploaded. The stream **must** be at the beginning (i.e. 

486 ``stream.tell() == 0``). 

487 metadata (Mapping[str, str]): The resource metadata, such as an 

488 ACL list. 

489 content_type (str): The content type of the resource, e.g. a JPEG 

490 image has content type ``image/jpeg``. 

491 total_bytes (Optional[int]): The total number of bytes to be 

492 uploaded. If specified, the upload size **will not** be 

493 determined from the stream (even if ``stream_final=True``). 

494 stream_final (Optional[bool]): Indicates if the ``stream`` is 

495 "final" (i.e. no more bytes will be added to it). In this case 

496 we determine the upload size from the size of the stream. If 

497 ``total_bytes`` is passed, this argument will be ignored. 

498 

499 Returns: 

500 Tuple[str, str, bytes, Mapping[str, str]]: The quadruple 

501 

502 * HTTP verb for the request (always POST) 

503 * the URL for the request 

504 * the body of the request 

505 * headers for the request 

506 

507 Raises: 

508 ValueError: If the current upload has already been initiated. 

509 ValueError: If ``stream`` is not at the beginning. 

510 

511 .. _sans-I/O: https://sans-io.readthedocs.io/ 

512 """ 

513 if self.resumable_url is not None: 

514 raise ValueError("This upload has already been initiated.") 

515 if stream.tell() != 0: 

516 raise ValueError("Stream must be at beginning.") 

517 

518 self._stream = stream 

519 self._content_type = content_type 

520 

521 # Signed URL requires content type set directly - not through x-upload-content-type 

522 parse_result = urllib.parse.urlparse(self.upload_url) 

523 parsed_query = urllib.parse.parse_qs(parse_result.query) 

524 if "x-goog-signature" in parsed_query or "X-Goog-Signature" in parsed_query: 

525 # Deconstruct **self._headers first so that content type defined here takes priority 

526 headers = {**self._headers, _CONTENT_TYPE_HEADER: content_type} 

527 else: 

528 # Deconstruct **self._headers first so that content type defined here takes priority 

529 headers = { 

530 **self._headers, 

531 _CONTENT_TYPE_HEADER: "application/json; charset=UTF-8", 

532 "x-upload-content-type": content_type, 

533 } 

534 # Set the total bytes if possible. 

535 if total_bytes is not None: 

536 self._total_bytes = total_bytes 

537 elif stream_final: 

538 self._total_bytes = get_total_bytes(stream) 

539 # Add the total bytes to the headers if set. 

540 if self._total_bytes is not None: 

541 content_length = "{:d}".format(self._total_bytes) 

542 headers["x-upload-content-length"] = content_length 

543 

544 payload = json.dumps(metadata).encode("utf-8") 

545 return _POST, self.upload_url, payload, headers 

546 

547 def _process_initiate_response(self, response): 

548 """Process the response from an HTTP request that initiated upload. 

549 

550 This is everything that must be done after a request that doesn't 

551 require network I/O (or other I/O). This is based on the `sans-I/O`_ 

552 philosophy. 

553 

554 This method takes the URL from the ``Location`` header and stores it 

555 for future use. Within that URL, we assume the ``upload_id`` query 

556 parameter has been included, but we do not check. 

557 

558 Args: 

559 response (object): The HTTP response object (need headers). 

560 

561 .. _sans-I/O: https://sans-io.readthedocs.io/ 

562 """ 

563 _helpers.require_status_code( 

564 response, 

565 (http.client.OK, http.client.CREATED), 

566 self._get_status_code, 

567 callback=self._make_invalid, 

568 ) 

569 self._resumable_url = _helpers.header_required( 

570 response, "location", self._get_headers 

571 ) 

572 

573 def initiate( 

574 self, 

575 transport, 

576 stream, 

577 metadata, 

578 content_type, 

579 total_bytes=None, 

580 stream_final=True, 

581 timeout=None, 

582 ): 

583 """Initiate a resumable upload. 

584 

585 By default, this method assumes your ``stream`` is in a "final" 

586 state ready to transmit. However, ``stream_final=False`` can be used 

587 to indicate that the size of the resource is not known. This can happen 

588 if bytes are being dynamically fed into ``stream``, e.g. if the stream 

589 is attached to application logs. 

590 

591 If ``stream_final=False`` is used, :attr:`chunk_size` bytes will be 

592 read from the stream every time :meth:`transmit_next_chunk` is called. 

593 If one of those reads produces strictly fewer bites than the chunk 

594 size, the upload will be concluded. 

595 

596 Args: 

597 transport (object): An object which can make authenticated 

598 requests. 

599 stream (IO[bytes]): The stream (i.e. file-like object) that will 

600 be uploaded. The stream **must** be at the beginning (i.e. 

601 ``stream.tell() == 0``). 

602 metadata (Mapping[str, str]): The resource metadata, such as an 

603 ACL list. 

604 content_type (str): The content type of the resource, e.g. a JPEG 

605 image has content type ``image/jpeg``. 

606 total_bytes (Optional[int]): The total number of bytes to be 

607 uploaded. If specified, the upload size **will not** be 

608 determined from the stream (even if ``stream_final=True``). 

609 stream_final (Optional[bool]): Indicates if the ``stream`` is 

610 "final" (i.e. no more bytes will be added to it). In this case 

611 we determine the upload size from the size of the stream. If 

612 ``total_bytes`` is passed, this argument will be ignored. 

613 timeout (Optional[Union[float, Tuple[float, float]]]): 

614 The number of seconds to wait for the server response. 

615 Depending on the retry strategy, a request may be repeated 

616 several times using the same timeout each time. 

617 

618 Can also be passed as a tuple (connect_timeout, read_timeout). 

619 See :meth:`requests.Session.request` documentation for details. 

620 

621 Raises: 

622 NotImplementedError: Always, since virtual. 

623 """ 

624 raise NotImplementedError("This implementation is virtual.") 

625 

626 def _prepare_request(self): 

627 """Prepare the contents of HTTP request to upload a chunk. 

628 

629 This is everything that must be done before a request that doesn't 

630 require network I/O. This is based on the `sans-I/O`_ philosophy. 

631 

632 For the time being, this **does require** some form of I/O to read 

633 a chunk from ``stream`` (via :func:`get_next_chunk`). However, this 

634 will (almost) certainly not be network I/O. 

635 

636 Returns: 

637 Tuple[str, str, bytes, Mapping[str, str]]: The quadruple 

638 

639 * HTTP verb for the request (always PUT) 

640 * the URL for the request 

641 * the body of the request 

642 * headers for the request 

643 

644 The headers incorporate the ``_headers`` on the current instance. 

645 

646 Raises: 

647 ValueError: If the current upload has finished. 

648 ValueError: If the current upload is in an invalid state. 

649 ValueError: If the current upload has not been initiated. 

650 ValueError: If the location in the stream (i.e. ``stream.tell()``) 

651 does not agree with ``bytes_uploaded``. 

652 

653 .. _sans-I/O: https://sans-io.readthedocs.io/ 

654 """ 

655 if self.finished: 

656 raise ValueError("Upload has finished.") 

657 if self.invalid: 

658 raise ValueError( 

659 "Upload is in an invalid state. To recover call `recover()`." 

660 ) 

661 if self.resumable_url is None: 

662 raise ValueError( 

663 "This upload has not been initiated. Please call " 

664 "initiate() before beginning to transmit chunks." 

665 ) 

666 

667 start_byte, payload, content_range = get_next_chunk( 

668 self._stream, self._chunk_size, self._total_bytes 

669 ) 

670 if start_byte != self.bytes_uploaded: 

671 msg = _STREAM_ERROR_TEMPLATE.format(start_byte, self.bytes_uploaded) 

672 raise ValueError(msg) 

673 

674 self._update_checksum(start_byte, payload) 

675 

676 headers = { 

677 **self._headers, 

678 _CONTENT_TYPE_HEADER: self._content_type, 

679 _helpers.CONTENT_RANGE_HEADER: content_range, 

680 } 

681 return _PUT, self.resumable_url, payload, headers 

682 

683 def _update_checksum(self, start_byte, payload): 

684 """Update the checksum with the payload if not already updated. 

685 

686 Because error recovery can result in bytes being transmitted more than 

687 once, the checksum tracks the number of bytes checked in 

688 self._bytes_checksummed and skips bytes that have already been summed. 

689 """ 

690 if not self._checksum_type: 

691 return 

692 

693 if not self._checksum_object: 

694 self._checksum_object = _helpers._get_checksum_object(self._checksum_type) 

695 

696 if start_byte < self._bytes_checksummed: 

697 offset = self._bytes_checksummed - start_byte 

698 data = payload[offset:] 

699 else: 

700 data = payload 

701 

702 self._checksum_object.update(data) 

703 self._bytes_checksummed += len(data) 

704 

705 def _make_invalid(self): 

706 """Simple setter for ``invalid``. 

707 

708 This is intended to be passed along as a callback to helpers that 

709 raise an exception so they can mark this instance as invalid before 

710 raising. 

711 """ 

712 self._invalid = True 

713 

714 def _process_resumable_response(self, response, bytes_sent): 

715 """Process the response from an HTTP request. 

716 

717 This is everything that must be done after a request that doesn't 

718 require network I/O (or other I/O). This is based on the `sans-I/O`_ 

719 philosophy. 

720 

721 Args: 

722 response (object): The HTTP response object. 

723 bytes_sent (int): The number of bytes sent in the request that 

724 ``response`` was returned for. 

725 

726 Raises: 

727 ~google.cloud.storage.exceptions.InvalidResponse: If the status 

728 code is 308 and the ``range`` header is not of the form 

729 ``bytes 0-{end}``. 

730 ~google.cloud.storage.exceptions.InvalidResponse: If the status 

731 code is not 200 or 308. 

732 

733 .. _sans-I/O: https://sans-io.readthedocs.io/ 

734 """ 

735 status_code = _helpers.require_status_code( 

736 response, 

737 (http.client.OK, http.client.PERMANENT_REDIRECT), 

738 self._get_status_code, 

739 callback=self._make_invalid, 

740 ) 

741 if status_code == http.client.OK: 

742 # NOTE: We use the "local" information of ``bytes_sent`` to update 

743 # ``bytes_uploaded``, but do not verify this against other 

744 # state. However, there may be some other information: 

745 # 

746 # * a ``size`` key in JSON response body 

747 # * the ``total_bytes`` attribute (if set) 

748 # * ``stream.tell()`` (relying on fact that ``initiate()`` 

749 # requires stream to be at the beginning) 

750 self._bytes_uploaded = self._bytes_uploaded + bytes_sent 

751 # Tombstone the current upload so it cannot be used again. 

752 self._finished = True 

753 # Validate the checksum. This can raise an exception on failure. 

754 self._validate_checksum(response) 

755 else: 

756 bytes_range = _helpers.header_required( 

757 response, 

758 _helpers.RANGE_HEADER, 

759 self._get_headers, 

760 callback=self._make_invalid, 

761 ) 

762 match = _BYTES_RANGE_RE.match(bytes_range) 

763 if match is None: 

764 self._make_invalid() 

765 raise InvalidResponse( 

766 response, 

767 'Unexpected "range" header', 

768 bytes_range, 

769 'Expected to be of the form "bytes=0-{end}"', 

770 ) 

771 self._bytes_uploaded = int(match.group("end_byte")) + 1 

772 

773 def _validate_checksum(self, response): 

774 """Check the computed checksum, if any, against the recieved metadata. 

775 

776 Args: 

777 response (object): The HTTP response object. 

778 

779 Raises: 

780 ~google.cloud.storage.exceptions.DataCorruption: If the checksum 

781 computed locally and the checksum reported by the remote host do 

782 not match. 

783 """ 

784 if self._checksum_type is None: 

785 return 

786 metadata_key = _helpers._get_metadata_key(self._checksum_type) 

787 metadata = response.json() 

788 remote_checksum = metadata.get(metadata_key) 

789 if remote_checksum is None: 

790 raise InvalidResponse( 

791 response, 

792 _UPLOAD_METADATA_NO_APPROPRIATE_CHECKSUM_MESSAGE.format(metadata_key), 

793 self._get_headers(response), 

794 ) 

795 local_checksum = _helpers.prepare_checksum_digest( 

796 self._checksum_object.digest() 

797 ) 

798 if local_checksum != remote_checksum: 

799 raise DataCorruption( 

800 response, 

801 _UPLOAD_CHECKSUM_MISMATCH_MESSAGE.format( 

802 self._checksum_type.upper(), local_checksum, remote_checksum 

803 ), 

804 ) 

805 

806 def transmit_next_chunk(self, transport, timeout=None): 

807 """Transmit the next chunk of the resource to be uploaded. 

808 

809 If the current upload was initiated with ``stream_final=False``, 

810 this method will dynamically determine if the upload has completed. 

811 The upload will be considered complete if the stream produces 

812 fewer than :attr:`chunk_size` bytes when a chunk is read from it. 

813 

814 Args: 

815 transport (object): An object which can make authenticated 

816 requests. 

817 timeout (Optional[Union[float, Tuple[float, float]]]): 

818 The number of seconds to wait for the server response. 

819 Depending on the retry strategy, a request may be repeated 

820 several times using the same timeout each time. 

821 

822 Can also be passed as a tuple (connect_timeout, read_timeout). 

823 See :meth:`requests.Session.request` documentation for details. 

824 

825 Raises: 

826 NotImplementedError: Always, since virtual. 

827 """ 

828 raise NotImplementedError("This implementation is virtual.") 

829 

830 def _prepare_recover_request(self): 

831 """Prepare the contents of HTTP request to recover from failure. 

832 

833 This is everything that must be done before a request that doesn't 

834 require network I/O. This is based on the `sans-I/O`_ philosophy. 

835 

836 We assume that the :attr:`resumable_url` is set (i.e. the only way 

837 the upload can end up :attr:`invalid` is if it has been initiated. 

838 

839 Returns: 

840 Tuple[str, str, NoneType, Mapping[str, str]]: The quadruple 

841 

842 * HTTP verb for the request (always PUT) 

843 * the URL for the request 

844 * the body of the request (always :data:`None`) 

845 * headers for the request 

846 

847 The headers **do not** incorporate the ``_headers`` on the 

848 current instance. 

849 

850 .. _sans-I/O: https://sans-io.readthedocs.io/ 

851 """ 

852 headers = {_helpers.CONTENT_RANGE_HEADER: "bytes */*"} 

853 return _PUT, self.resumable_url, None, headers 

854 

855 def _process_recover_response(self, response): 

856 """Process the response from an HTTP request to recover from failure. 

857 

858 This is everything that must be done after a request that doesn't 

859 require network I/O (or other I/O). This is based on the `sans-I/O`_ 

860 philosophy. 

861 

862 Args: 

863 response (object): The HTTP response object. 

864 

865 Raises: 

866 ~google.cloud.storage.exceptions.InvalidResponse: If the status 

867 code is not 308. 

868 ~google.cloud.storage.exceptions.InvalidResponse: If the status 

869 code is 308 and the ``range`` header is not of the form 

870 ``bytes 0-{end}``. 

871 

872 .. _sans-I/O: https://sans-io.readthedocs.io/ 

873 """ 

874 _helpers.require_status_code( 

875 response, (http.client.PERMANENT_REDIRECT,), self._get_status_code 

876 ) 

877 headers = self._get_headers(response) 

878 if _helpers.RANGE_HEADER in headers: 

879 bytes_range = headers[_helpers.RANGE_HEADER] 

880 match = _BYTES_RANGE_RE.match(bytes_range) 

881 if match is None: 

882 raise InvalidResponse( 

883 response, 

884 'Unexpected "range" header', 

885 bytes_range, 

886 'Expected to be of the form "bytes=0-{end}"', 

887 ) 

888 self._bytes_uploaded = int(match.group("end_byte")) + 1 

889 else: 

890 # In this case, the upload has not "begun". 

891 self._bytes_uploaded = 0 

892 

893 self._stream.seek(self._bytes_uploaded) 

894 self._invalid = False 

895 

896 def recover(self, transport): 

897 """Recover from a failure. 

898 

899 This method should be used when a :class:`ResumableUpload` is in an 

900 :attr:`~ResumableUpload.invalid` state due to a request failure. 

901 

902 This will verify the progress with the server and make sure the 

903 current upload is in a valid state before :meth:`transmit_next_chunk` 

904 can be used again. 

905 

906 Args: 

907 transport (object): An object which can make authenticated 

908 requests. 

909 

910 Raises: 

911 NotImplementedError: Always, since virtual. 

912 """ 

913 raise NotImplementedError("This implementation is virtual.") 

914 

915 

916class XMLMPUContainer(UploadBase): 

917 """Initiate and close an upload using the XML MPU API. 

918 

919 An XML MPU sends an initial request and then receives an upload ID. 

920 Using the upload ID, the upload is then done in numbered parts and the 

921 parts can be uploaded concurrently. 

922 

923 In order to avoid concurrency issues with this container object, the 

924 uploading of individual parts is handled separately, by XMLMPUPart objects 

925 spawned from this container class. The XMLMPUPart objects are not 

926 necessarily in the same process as the container, so they do not update the 

927 container automatically. 

928 

929 MPUs are sometimes referred to as "Multipart Uploads", which is ambiguous 

930 given the JSON multipart upload, so the abbreviation "MPU" will be used 

931 throughout. 

932 

933 See: https://cloud.google.com/storage/docs/multipart-uploads 

934 

935 Args: 

936 upload_url (str): The URL of the object (without query parameters). The 

937 initiate, PUT, and finalization requests will all use this URL, with 

938 varying query parameters. 

939 filename (str): The name (path) of the file to upload. 

940 headers (Optional[Mapping[str, str]]): Extra headers that should 

941 be sent with every request. 

942 retry (Optional[google.api_core.retry.Retry]): How to retry the 

943 RPC. A None value will disable retries. A 

944 google.api_core.retry.Retry value will enable retries, and the 

945 object will configure backoff and timeout options. 

946 

947 See the retry.py source code and docstrings in this package 

948 (google.cloud.storage.retry) for information on retry types and how 

949 to configure them. 

950 

951 Attributes: 

952 upload_url (str): The URL where the content will be uploaded. 

953 upload_id (Optional(str)): The ID of the upload from the initialization 

954 response. 

955 """ 

956 

957 def __init__( 

958 self, upload_url, filename, headers=None, upload_id=None, retry=DEFAULT_RETRY 

959 ): 

960 super().__init__(upload_url, headers=headers, retry=retry) 

961 self._filename = filename 

962 self._upload_id = upload_id 

963 self._parts = {} 

964 

965 @property 

966 def upload_id(self): 

967 return self._upload_id 

968 

969 def register_part(self, part_number, etag): 

970 """Register an uploaded part by part number and corresponding etag. 

971 

972 XMLMPUPart objects represent individual parts, and their part number 

973 and etag can be registered to the container object with this method 

974 and therefore incorporated in the finalize() call to finish the upload. 

975 

976 This method accepts part_number and etag, but not XMLMPUPart objects 

977 themselves, to reduce the complexity involved in running XMLMPUPart 

978 uploads in separate processes. 

979 

980 Args: 

981 part_number (int): The part number. Parts are assembled into the 

982 final uploaded object with finalize() in order of their part 

983 numbers. 

984 etag (str): The etag included in the server response after upload. 

985 """ 

986 self._parts[part_number] = etag 

987 

988 def _prepare_initiate_request(self, content_type): 

989 """Prepare the contents of HTTP request to initiate upload. 

990 

991 This is everything that must be done before a request that doesn't 

992 require network I/O (or other I/O). This is based on the `sans-I/O`_ 

993 philosophy. 

994 

995 Args: 

996 content_type (str): The content type of the resource, e.g. a JPEG 

997 image has content type ``image/jpeg``. 

998 

999 Returns: 

1000 Tuple[str, str, bytes, Mapping[str, str]]: The quadruple 

1001 

1002 * HTTP verb for the request (always POST) 

1003 * the URL for the request 

1004 * the body of the request 

1005 * headers for the request 

1006 

1007 Raises: 

1008 ValueError: If the current upload has already been initiated. 

1009 

1010 .. _sans-I/O: https://sans-io.readthedocs.io/ 

1011 """ 

1012 if self.upload_id is not None: 

1013 raise ValueError("This upload has already been initiated.") 

1014 

1015 initiate_url = self.upload_url + _MPU_INITIATE_QUERY 

1016 

1017 headers = { 

1018 **self._headers, 

1019 _CONTENT_TYPE_HEADER: content_type, 

1020 } 

1021 return _POST, initiate_url, None, headers 

1022 

1023 def _process_initiate_response(self, response): 

1024 """Process the response from an HTTP request that initiated the upload. 

1025 

1026 This is everything that must be done after a request that doesn't 

1027 require network I/O (or other I/O). This is based on the `sans-I/O`_ 

1028 philosophy. 

1029 

1030 This method takes the URL from the ``Location`` header and stores it 

1031 for future use. Within that URL, we assume the ``upload_id`` query 

1032 parameter has been included, but we do not check. 

1033 

1034 Args: 

1035 response (object): The HTTP response object. 

1036 

1037 Raises: 

1038 ~google.cloud.storage.exceptions.InvalidResponse: If the status 

1039 code is not 200. 

1040 

1041 .. _sans-I/O: https://sans-io.readthedocs.io/ 

1042 """ 

1043 _helpers.require_status_code(response, (http.client.OK,), self._get_status_code) 

1044 root = ElementTree.fromstring(response.text) 

1045 self._upload_id = root.find(_S3_COMPAT_XML_NAMESPACE + _UPLOAD_ID_NODE).text 

1046 

1047 def initiate( 

1048 self, 

1049 transport, 

1050 content_type, 

1051 timeout=None, 

1052 ): 

1053 """Initiate an MPU and record the upload ID. 

1054 

1055 Args: 

1056 transport (object): An object which can make authenticated 

1057 requests. 

1058 content_type (str): The content type of the resource, e.g. a JPEG 

1059 image has content type ``image/jpeg``. 

1060 timeout (Optional[Union[float, Tuple[float, float]]]): 

1061 The number of seconds to wait for the server response. 

1062 Depending on the retry strategy, a request may be repeated 

1063 several times using the same timeout each time. 

1064 

1065 Can also be passed as a tuple (connect_timeout, read_timeout). 

1066 See :meth:`requests.Session.request` documentation for details. 

1067 

1068 Raises: 

1069 NotImplementedError: Always, since virtual. 

1070 """ 

1071 raise NotImplementedError("This implementation is virtual.") 

1072 

1073 def _prepare_finalize_request(self): 

1074 """Prepare the contents of an HTTP request to finalize the upload. 

1075 

1076 All of the parts must be registered before calling this method. 

1077 

1078 Returns: 

1079 Tuple[str, str, bytes, Mapping[str, str]]: The quadruple 

1080 

1081 * HTTP verb for the request (always POST) 

1082 * the URL for the request 

1083 * the body of the request 

1084 * headers for the request 

1085 

1086 Raises: 

1087 ValueError: If the upload has not been initiated. 

1088 """ 

1089 if self.upload_id is None: 

1090 raise ValueError("This upload has not yet been initiated.") 

1091 

1092 final_query = _MPU_FINAL_QUERY_TEMPLATE.format(upload_id=self._upload_id) 

1093 finalize_url = self.upload_url + final_query 

1094 final_xml_root = ElementTree.Element("CompleteMultipartUpload")