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 15"""Classes to encapsulate a single HTTP request. 16 17The classes implement a command pattern, with every 18object supporting an execute() method that does the 19actual HTTP request. 20""" 21from __future__ import absolute_import 22 23__author__ = "[email protected] (Joe Gregorio)" 24 25import copy 26import httplib2 27import http.client as http_client 28import io 29import json 30import logging 31import mimetypes 32import os 33import random 34import socket 35import time 36import urllib 37import uuid 38 39# TODO(issue 221): Remove this conditional import jibbajabba. 40try: 41 import ssl 42except ImportError: 43 _ssl_SSLError = object() 44else: 45 _ssl_SSLError = ssl.SSLError 46 47from email.generator import Generator 48from email.mime.multipart import MIMEMultipart 49from email.mime.nonmultipart import MIMENonMultipart 50from email.parser import FeedParser 51 52from googleapiclient import _helpers as util 53 54from googleapiclient import _auth 55from googleapiclient.errors import BatchError 56from googleapiclient.errors import HttpError 57from googleapiclient.errors import InvalidChunkSizeError 58from googleapiclient.errors import ResumableUploadError 59from googleapiclient.errors import UnexpectedBodyError 60from googleapiclient.errors import UnexpectedMethodError 61from googleapiclient.model import JsonModel 62 63 64LOGGER = logging.getLogger(__name__) 65 66DEFAULT_CHUNK_SIZE = 100 * 1024 * 1024 67 68MAX_URI_LENGTH = 2048 69 70MAX_BATCH_LIMIT = 1000 71 72_TOO_MANY_REQUESTS = 429 73 74DEFAULT_HTTP_TIMEOUT_SEC = 60 75 76_LEGACY_BATCH_URI = "https://www.googleapis.com/batch" 77 78 79def _should_retry_response(resp_status, content): 80 """Determines whether a response should be retried. 81 82 Args: 83 resp_status: The response status received. 84 content: The response content body. 85 86 Returns: 87 True if the response should be retried, otherwise False. 88 """ 89 reason = None 90 91 # Retry on 5xx errors. 92 if resp_status >= 500: 93 return True 94 95 # Retry on 429 errors. 96 if resp_status == _TOO_MANY_REQUESTS: 97 return True 98 99 # For 403 errors, we have to check for the `reason` in the response to 100 # determine if we should retry. 101 if resp_status == http_client.FORBIDDEN: 102 # If there's no details about the 403 type, don't retry. 103 if not content: 104 return False 105 106 # Content is in JSON format. 107 try: 108 data = json.loads(content.decode("utf-8")) 109 if isinstance(data, dict): 110 # There are many variations of the error json so we need 111 # to determine the keyword which has the error detail. Make sure 112 # that the order of the keywords below isn't changed as it can 113 # break user code. If the "errors" key exists, we must use that 114 # first. 115 # See Issue #1243 116 # https://github.com/googleapis/google-api-python-client/issues/1243 117 error_detail_keyword = next( 118 ( 119 kw 120 for kw in ["errors", "status", "message"] 121 if kw in data["error"] 122 ), 123 "", 124 ) 125 126 if error_detail_keyword: 127 reason = data["error"][error_detail_keyword] 128 129 if isinstance(reason, list) and len(reason) > 0: 130 reason = reason[0] 131 if "reason" in reason: 132 reason = reason["reason"] 133 else: 134 reason = data[0]["error"]["errors"]["reason"] 135 except (UnicodeDecodeError, ValueError, KeyError): 136 LOGGER.warning("Invalid JSON content from response: %s", content) 137 return False 138 139 LOGGER.warning('Encountered 403 Forbidden with reason "%s"', reason) 140 141 # Only retry on rate limit related failures. 142 if reason in ("userRateLimitExceeded", "rateLimitExceeded"): 143 return True 144 145 # Everything else is a success or non-retriable so break. 146 return False 147 148 149def _retry_request( 150 http, num_retries, req_type, sleep, rand, uri, method, *args, **kwargs 151): 152 """Retries an HTTP request multiple times while handling errors. 153 154 If after all retries the request still fails, last error is either returned as 155 return value (for HTTP 5xx errors) or thrown (for ssl.SSLError). 156 157 Args: 158 http: Http object to be used to execute request. 159 num_retries: Maximum number of retries. 160 req_type: Type of the request (used for logging retries). 161 sleep, rand: Functions to sleep for random time between retries. 162 uri: URI to be requested. 163 method: HTTP method to be used. 164 args, kwargs: Additional arguments passed to http.request. 165 166 Returns: 167 resp, content - Response from the http request (may be HTTP 5xx). 168 """ 169 resp = None 170 content = None 171 exception = None 172 for retry_num in range(num_retries + 1): 173 if retry_num > 0: 174 # Sleep before retrying. 175 sleep_time = rand() * 2 ** retry_num 176 LOGGER.warning( 177 "Sleeping %.2f seconds before retry %d of %d for %s: %s %s, after %s", 178 sleep_time, 179 retry_num, 180 num_retries, 181 req_type, 182 method, 183 uri, 184 resp.status if resp else exception, 185 ) 186 sleep(sleep_time) 187 188 try: 189 exception = None 190 resp, content = http.request(uri, method, *args, **kwargs) 191 # Retry on SSL errors and socket timeout errors. 192 except _ssl_SSLError as ssl_error: 193 exception = ssl_error 194 except socket.timeout as socket_timeout: 195 # Needs to be before socket.error as it's a subclass of OSError 196 # socket.timeout has no errorcode 197 exception = socket_timeout 198 except ConnectionError as connection_error: 199 # Needs to be before socket.error as it's a subclass of OSError 200 exception = connection_error 201 except OSError as socket_error: 202 # errno's contents differ by platform, so we have to match by name. 203 # Some of these same errors may have been caught above, e.g. ECONNRESET *should* be 204 # raised as a ConnectionError, but some libraries will raise it as a socket.error 205 # with an errno corresponding to ECONNRESET 206 if socket.errno.errorcode.get(socket_error.errno) not in { 207 "WSAETIMEDOUT", 208 "ETIMEDOUT", 209 "EPIPE", 210 "ECONNABORTED", 211 "ECONNREFUSED", 212 "ECONNRESET", 213 }: 214 raise 215 exception = socket_error 216 except httplib2.ServerNotFoundError as server_not_found_error: 217 exception = server_not_found_error 218 219 if exception: 220 if retry_num == num_retries: 221 raise exception 222 else: 223 continue 224 225 if not _should_retry_response(resp.status, content): 226 break 227 228 return resp, content 229 230 231class MediaUploadProgress(object): 232 """Status of a resumable upload.""" 233 234 def __init__(self, resumable_progress, total_size): 235 """Constructor. 236 237 Args: 238 resumable_progress: int, bytes sent so far. 239 total_size: int, total bytes in complete upload, or None if the total 240 upload size isn't known ahead of time. 241 """ 242 self.resumable_progress = resumable_progress 243 self.total_size = total_size 244 245 def progress(self): 246 """Percent of upload completed, as a float. 247 248 Returns: 249 the percentage complete as a float, returning 0.0 if the total size of 250 the upload is unknown. 251 """ 252 if self.total_size is not None and self.total_size != 0: 253 return float(self.resumable_progress) / float(self.total_size) 254 else: 255 return 0.0 256 257 258class MediaDownloadProgress(object): 259 """Status of a resumable download.""" 260 261 def __init__(self, resumable_progress, total_size): 262 """Constructor. 263 264 Args: 265 resumable_progress: int, bytes received so far. 266 total_size: int, total bytes in complete download. 267 """ 268 self.resumable_progress = resumable_progress 269 self.total_size = total_size 270 271 def progress(self): 272 """Percent of download completed, as a float. 273 274 Returns: 275 the percentage complete as a float, returning 0.0 if the total size of 276 the download is unknown. 277 """ 278 if self.total_size is not None and self.total_size != 0: 279 return float(self.resumable_progress) / float(self.total_size) 280 else: 281 return 0.0 282 283 284class MediaUpload(object): 285 """Describes a media object to upload. 286 287 Base class that defines the interface of MediaUpload subclasses. 288 289 Note that subclasses of MediaUpload may allow you to control the chunksize 290 when uploading a media object. It is important to keep the size of the chunk 291 as large as possible to keep the upload efficient. Other factors may influence 292 the size of the chunk you use, particularly if you are working in an 293 environment where individual HTTP requests may have a hardcoded time limit, 294 such as under certain classes of requests under Google App Engine. 295 296 Streams are io.Base compatible objects that support seek(). Some MediaUpload 297 subclasses support using streams directly to upload data. Support for 298 streaming may be indicated by a MediaUpload sub-class and if appropriate for a 299 platform that stream will be used for uploading the media object. The support 300 for streaming is indicated by has_stream() returning True. The stream() method 301 should return an io.Base object that supports seek(). On platforms where the 302 underlying httplib module supports streaming, for example Python 2.6 and 303 later, the stream will be passed into the http library which will result in 304 less memory being used and possibly faster uploads. 305 306 If you need to upload media that can't be uploaded using any of the existing 307 MediaUpload sub-class then you can sub-class MediaUpload for your particular 308 needs. 309 """ 310 311 def chunksize(self): 312 """Chunk size for resumable uploads. 313 314 Returns: 315 Chunk size in bytes. 316 """ 317 raise NotImplementedError() 318 319 def mimetype(self): 320 """Mime type of the body. 321 322 Returns: 323 Mime type. 324 """ 325 return "application/octet-stream" 326 327 def size(self): 328 """Size of upload. 329 330 Returns: 331 Size of the body, or None of the size is unknown. 332 """ 333 return None 334 335 def resumable(self): 336 """Whether this upload is resumable. 337 338 Returns: 339 True if resumable upload or False. 340 """ 341 return False 342 343 def getbytes(self, begin, end): 344 """Get bytes from the media. 345 346 Args: 347 begin: int, offset from beginning of file. 348 length: int, number of bytes to read, starting at begin. 349 350 Returns: 351 A string of bytes read. May be shorter than length if EOF was reached 352 first. 353 """ 354 raise NotImplementedError() 355 356 def has_stream(self): 357 """Does the underlying upload support a streaming interface. 358 359 Streaming means it is an io.IOBase subclass that supports seek, i.e. 360 seekable() returns True. 361 362 Returns: 363 True if the call to stream() will return an instance of a seekable io.Base 364 subclass. 365 """ 366 return False 367 368 def stream(self): 369 """A stream interface to the data being uploaded. 370 371 Returns: 372 The returned value is an io.IOBase subclass that supports seek, i.e. 373 seekable() returns True. 374 """ 375 raise NotImplementedError() 376 377 @util.positional(1) 378 def _to_json(self, strip=None): 379 """Utility function for creating a JSON representation of a MediaUpload. 380 381 Args: 382 strip: array, An array of names of members to not include in the JSON. 383 384 Returns: 385 string, a JSON representation of this instance, suitable to pass to 386 from_json(). 387 """ 388 t = type(self) 389 d = copy.copy(self.__dict__) 390 if strip is not None: 391 for member in strip: 392 del d[member] 393 d["_class"] = t.__name__ 394 d["_module"] = t.__module__ 395 return json.dumps(d) 396 397 def to_json(self): 398 """Create a JSON representation of an instance of MediaUpload. 399 400 Returns: 401 string, a JSON representation of this instance, suitable to pass to 402 from_json(). 403 """ 404 return self._to_json() 405 406 @classmethod 407 def new_from_json(cls, s): 408 """Utility class method to instantiate a MediaUpload subclass from a JSON 409 representation produced by to_json(). 410 411 Args: 412 s: string, JSON from to_json(). 413 414 Returns: 415 An instance of the subclass of MediaUpload that was serialized with 416 to_json(). 417 """ 418 data = json.loads(s) 419 # Find and call the right classmethod from_json() to restore the object. 420 module = data["_module"] 421 m = __import__(module, fromlist=module.split(".")[:-1]) 422 kls = getattr(m, data["_class"]) 423 from_json = getattr(kls, "from_json") 424 return from_json(s) 425 426 427class MediaIoBaseUpload(MediaUpload): 428 """A MediaUpload for a io.Base objects. 429 430 Note that the Python file object is compatible with io.Base and can be used 431 with this class also. 432 433 fh = BytesIO('...Some data to upload...') 434 media = MediaIoBaseUpload(fh, mimetype='image/png', 435 chunksize=1024*1024, resumable=True) 436 farm.animals().insert( 437 id='cow', 438 name='cow.png', 439 media_body=media).execute() 440 441 Depending on the platform you are working on, you may pass -1 as the 442 chunksize, which indicates that the entire file should be uploaded in a single 443 request. If the underlying platform supports streams, such as Python 2.6 or 444 later, then this can be very efficient as it avoids multiple connections, and 445 also avoids loading the entire file into memory before sending it. Note that 446 Google App Engine has a 5MB limit on request size, so you should never set 447 your chunksize larger than 5MB, or to -1. 448 """ 449 450 @util.positional(3) 451 def __init__(self, fd, mimetype, chunksize=DEFAULT_CHUNK_SIZE, resumable=False): 452 """Constructor. 453 454 Args: 455 fd: io.Base or file object, The source of the bytes to upload. MUST be 456 opened in blocking mode, do not use streams opened in non-blocking mode. 457 The given stream must be seekable, that is, it must be able to call 458 seek() on fd. 459 mimetype: string, Mime-type of the file. 460 chunksize: int, File will be uploaded in chunks of this many bytes. Only 461 used if resumable=True. Pass in a value of -1 if the file is to be 462 uploaded as a single chunk. Note that Google App Engine has a 5MB limit 463 on request size, so you should never set your chunksize larger than 5MB, 464 or to -1. 465 resumable: bool, True if this is a resumable upload. False means upload 466 in a single request. 467 """ 468 super(MediaIoBaseUpload, self).__init__() 469 self._fd = fd 470 self._mimetype = mimetype 471 if not (chunksize == -1 or chunksize > 0): 472 raise InvalidChunkSizeError() 473 self._chunksize = chunksize 474 self._resumable = resumable 475 476 self._fd.seek(0, os.SEEK_END) 477 self._size = self._fd.tell() 478 479 def chunksize(self): 480 """Chunk size for resumable uploads. 481 482 Returns: 483 Chunk size in bytes. 484 """ 485 return self._chunksize 486 487 def mimetype(self): 488 """Mime type of the body. 489 490 Returns: 491 Mime type. 492 """ 493 return self._mimetype 494 495 def size(self): 496 """Size of upload. 497 498 Returns: 499 Size of the body, or None of the size is unknown. 500 """ 501 return self._size 502 503 def resumable(self): 504 """Whether this upload is resumable. 505 506 Returns: 507 True if resumable upload or False. 508 """ 509 return self._resumable 510 511 def getbytes(self, begin, length): 512 """Get bytes from the media. 513 514 Args: 515 begin: int, offset from beginning of file. 516 length: int, number of bytes to read, starting at begin. 517 518 Returns: 519 A string of bytes read. May be shorted than length if EOF was reached 520 first. 521 """ 522 self._fd.seek(begin) 523 return self._fd.read(length) 524 525 def has_stream(self): 526 """Does the underlying upload support a streaming interface. 527 528 Streaming means it is an io.IOBase subclass that supports seek, i.e. 529 seekable() returns True. 530 531 Returns: 532 True if the call to stream() will return an instance of a seekable io.Base 533 subclass. 534 """ 535 return True 536 537 def stream(self): 538 """A stream interface to the data being uploaded. 539 540 Returns: 541 The returned value is an io.IOBase subclass that supports seek, i.e. 542 seekable() returns True. 543 """ 544 return self._fd 545 546 def to_json(self): 547 """This upload type is not serializable.""" 548 raise NotImplementedError("MediaIoBaseUpload is not serializable.") 549 550 551class MediaFileUpload(MediaIoBaseUpload): 552 """A MediaUpload for a file. 553 554 Construct a MediaFileUpload and pass as the media_body parameter of the 555 method. For example, if we had a service that allowed uploading images: 556 557 media = MediaFileUpload('cow.png', mimetype='image/png', 558 chunksize=1024*1024, resumable=True) 559 farm.animals().insert( 560 id='cow', 561 name='cow.png', 562 media_body=media).execute() 563 564 Depending on the platform you are working on, you may pass -1 as the 565 chunksize, which indicates that the entire file should be uploaded in a single 566 request. If the underlying platform supports streams, such as Python 2.6 or 567 later, then this can be very efficient as it avoids multiple connections, and 568 also avoids loading the entire file into memory before sending it. Note that 569 Google App Engine has a 5MB limit on request size, so you should never set 570 your chunksize larger than 5MB, or to -1. 571 """ 572 573 @util.positional(2) 574 def __init__( 575 self, filename, mimetype=None, chunksize=DEFAULT_CHUNK_SIZE, resumable=False 576 ): 577 """Constructor. 578 579 Args: 580 filename: string, Name of the file. 581 mimetype: string, Mime-type of the file. If None then a mime-type will be 582 guessed from the file extension. 583 chunksize: int, File will be uploaded in chunks of this many bytes. Only 584 used if resumable=True. Pass in a value of -1 if the file is to be 585 uploaded in a single chunk. Note that Google App Engine has a 5MB limit 586 on request size, so you should never set your chunksize larger than 5MB, 587 or to -1. 588 resumable: bool, True if this is a resumable upload. False means upload 589 in a single request. 590 """ 591 self._fd = None 592 self._filename = filename 593 self._fd = open(self._filename, "rb") 594 if mimetype is None: 595 # No mimetype provided, make a guess. 596 mimetype, _ = mimetypes.guess_type(filename) 597 if mimetype is None: 598 # Guess failed, use octet-stream. 599 mimetype = "application/octet-stream" 600 super(MediaFileUpload, self).__init__( 601 self._fd, mimetype, chunksize=chunksize, resumable=resumable 602 ) 603 604 def __del__(self): 605 if self._fd: 606 self._fd.close() 607 608 def to_json(self): 609 """Creating a JSON representation of an instance of MediaFileUpload. 610 611 Returns: 612 string, a JSON representation of this instance, suitable to pass to 613 from_json(). 614 """ 615 return self._to_json(strip=["_fd"]) 616 617 @staticmethod 618 def from_json(s): 619 d = json.loads(s) 620 return MediaFileUpload( 621 d["_filename"], 622 mimetype=d["_mimetype"], 623 chunksize=d["_chunksize"], 624 resumable=d["_resumable"], 625 ) 626 627 628class MediaInMemoryUpload(MediaIoBaseUpload): 629 """MediaUpload for a chunk of bytes. 630 631 DEPRECATED: Use MediaIoBaseUpload with either io.TextIOBase or io.StringIO for 632 the stream. 633 """ 634 635 @util.positional(2) 636 def __init__( 637 self, 638 body, 639 mimetype="application/octet-stream", 640 chunksize=DEFAULT_CHUNK_SIZE, 641 resumable=False, 642 ): 643 """Create a new MediaInMemoryUpload. 644 645 DEPRECATED: Use MediaIoBaseUpload with either io.TextIOBase or io.StringIO for 646 the stream. 647 648 Args: 649 body: string, Bytes of body content. 650 mimetype: string, Mime-type of the file or default of 651 'application/octet-stream'. 652 chunksize: int, File will be uploaded in chunks of this many bytes. Only 653 used if resumable=True. 654 resumable: bool, True if this is a resumable upload. False means upload 655 in a single request. 656 """ 657 fd = io.BytesIO(body) 658 super(MediaInMemoryUpload, self).__init__( 659 fd, mimetype, chunksize=chunksize, resumable=resumable 660 ) 661 662 663class MediaIoBaseDownload(object): 664 """ "Download media resources. 665 666 Note that the Python file object is compatible with io.Base and can be used 667 with this class also. 668 669 670 Example: 671 request = farms.animals().get_media(id='cow') 672 fh = io.FileIO('cow.png', mode='wb') 673 downloader = MediaIoBaseDownload(fh, request, chunksize=1024*1024) 674 675 done = False 676 while done is False: 677 status, done = downloader.next_chunk() 678 if status: 679 print "Download %d%%." % int(status.progress() * 100) 680 print "Download Complete!" 681 """ 682 683 @util.positional(3) 684 def __init__(self, fd, request, chunksize=DEFAULT_CHUNK_SIZE): 685 """Constructor. 686 687 Args: 688 fd: io.Base or file object, The stream in which to write the downloaded 689 bytes. 690 request: googleapiclient.http.HttpRequest, the media request to perform in 691 chunks. 692 chunksize: int, File will be downloaded in chunks of this many bytes. 693 """ 694 self._fd = fd 695 self._request = request 696 self._uri = request.uri 697 self._chunksize = chunksize 698 self._progress = 0 699 self._total_size = None 700 self._done = False 701 702 # Stubs for testing. 703 self._sleep = time.sleep 704 self._rand = random.random 705 706 self._headers = {} 707 for k, v in request.headers.items(): 708 # allow users to supply custom headers by setting them on the request 709 # but strip out the ones that are set by default on requests generated by 710 # API methods like Drive's files().get(fileId=...) 711 if not k.lower() in ("accept", "accept-encoding", "user-agent"): 712 self._headers[k] = v 713 714 @util.positional(1) 715 def next_chunk(self, num_retries=0): 716 """Get the next chunk of the download. 717 718 Args: 719 num_retries: Integer, number of times to retry with randomized 720 exponential backoff. If all retries fail, the raised HttpError 721 represents the last request. If zero (default), we attempt the 722 request only once. 723 724 Returns: 725 (status, done): (MediaDownloadProgress, boolean) 726 The value of 'done' will be True when the media has been fully 727 downloaded or the total size of the media is unknown. 728 729 Raises: 730 googleapiclient.errors.HttpError if the response was not a 2xx. 731 httplib2.HttpLib2Error if a transport error has occurred. 732 """ 733 headers = self._headers.copy() 734 headers["range"] = "bytes=%d-%d" % ( 735 self._progress, 736 self._progress + self._chunksize - 1, 737 ) 738 http = self._request.http 739 740 resp, content = _retry_request( 741 http, 742 num_retries, 743 "media download", 744 self._sleep, 745 self._rand, 746 self._uri, 747 "GET", 748 headers=headers, 749 ) 750 751 if resp.status in [200, 206]: 752 if "content-location" in resp and resp["content-location"] != self._uri: 753 self._uri = resp["content-location"] 754 self._progress += len(content) 755 self._fd.write(content) 756 757 if "content-range" in resp: 758 content_range = resp["content-range"] 759 length = content_range.rsplit("/", 1)[1] 760 self._total_size = int(length) 761 elif "content-length" in resp: 762 self._total_size = int(resp["content-length"]) 763 764 if self._total_size is None or self._progress == self._total_size: 765 self._done = True 766 return MediaDownloadProgress(self._progress, self._total_size), self._done 767 elif resp.status == 416: 768 # 416 is Range Not Satisfiable 769 # This typically occurs with a zero byte file 770 content_range = resp["content-range"] 771 length = content_range.rsplit("/", 1)[1] 772 self._total_size = int(length) 773 if self._total_size == 0: 774 self._done = True 775 return ( 776 MediaDownloadProgress(self._progress, self._total_size), 777 self._done, 778 ) 779 raise HttpError(resp, content, uri=self._uri) 780 781 782class _StreamSlice(object): 783 """Truncated stream. 784 785 Takes a stream and presents a stream that is a slice of the original stream. 786 This is used when uploading media in chunks. In later versions of Python a 787 stream can be passed to httplib in place of the string of data to send. The 788 problem is that httplib just blindly reads to the end of the stream. This 789 wrapper presents a virtual stream that only reads to the end of the chunk. 790 """ 791 792 def __init__(self, stream, begin, chunksize): 793 """Constructor. 794 795 Args: 796 stream: (io.Base, file object), the stream to wrap. 797 begin: int, the seek position the chunk begins at. 798 chunksize: int, the size of the chunk. 799 """ 800 self._stream = stream 801 self._begin = begin 802 self._chunksize = chunksize 803 self._stream.seek(begin) 804 805 def read(self, n=-1): 806 """Read n bytes. 807 808 Args: 809 n, int, the number of bytes to read. 810 811 Returns: 812 A string of length 'n', or less if EOF is reached. 813 """ 814 # The data left available to read sits in [cur, end) 815 cur = self._stream.tell() 816 end = self._begin + self._chunksize 817 if n == -1 or cur + n > end: 818 n = end - cur 819 return self._stream.read(n) 820 821 822class HttpRequest(object): 823 """Encapsulates a single HTTP request.""" 824 825 @util.positional(4) 826 def __init__( 827 self, 828 http, 829 postproc, 830 uri, 831 method="GET", 832 body=None, 833 headers=None, 834 methodId=None, 835 resumable=None, 836 ): 837 """Constructor for an HttpRequest. 838 839 Args: 840 http: httplib2.Http, the transport object to use to make a request 841 postproc: callable, called on the HTTP response and content to transform 842 it into a data object before returning, or raising an exception 843 on an error. 844 uri: string, the absolute URI to send the request to 845 method: string, the HTTP method to use 846 body: string, the request body of the HTTP request, 847 headers: dict, the HTTP request headers 848 methodId: string, a unique identifier for the API method being called. 849 resumable: MediaUpload, None if this is not a resumbale request. 850 """ 851 self.uri = uri 852 self.method = method 853 self.body = body 854 self.headers = headers or {} 855 self.methodId = methodId 856 self.http = http 857 self.postproc = postproc 858 self.resumable = resumable 859 self.response_callbacks = [] 860 self._in_error_state = False 861 862 # The size of the non-media part of the request. 863 self.body_size = len(self.body or "") 864 865 # The resumable URI to send chunks to. 866 self.resumable_uri = None 867 868 # The bytes that have been uploaded. 869 self.resumable_progress = 0 870 871 # Stubs for testing. 872 self._rand = random.random 873 self._sleep = time.sleep 874 875 @util.positional(1) 876 def execute(self, http=None, num_retries=0): 877 """Execute the request. 878 879 Args: 880 http: httplib2.Http, an http object to be used in place of the 881 one the HttpRequest request object was constructed with. 882 num_retries: Integer, number of times to retry with randomized 883 exponential backoff. If all retries fail, the raised HttpError 884 represents the last request. If zero (default), we attempt the 885 request only once. 886 887 Returns: 888 A deserialized object model of the response body as determined 889 by the postproc. 890 891 Raises: 892 googleapiclient.errors.HttpError if the response was not a 2xx. 893 httplib2.HttpLib2Error if a transport error has occurred. 894 """ 895 if http is None: 896 http = self.http 897 898 if self.resumable: 899 body = None 900 while body is None: 901 _, body = self.next_chunk(http=http, num_retries=num_retries) 902 return body 903 904 # Non-resumable case. 905 906 if "content-length" not in self.headers: 907 self.headers["content-length"] = str(self.body_size) 908 # If the request URI is too long then turn it into a POST request. 909 # Assume that a GET request never contains a request body. 910 if len(self.uri) > MAX_URI_LENGTH and self.method == "GET": 911 self.method = "POST" 912 self.headers["x-http-method-override"] = "GET" 913 self.headers["content-type"] = "application/x-www-form-urlencoded" 914 parsed = urllib.parse.urlparse(self.uri) 915 self.uri = urllib.parse.urlunparse( 916 (parsed.scheme, parsed.netloc, parsed.path, parsed.params, None, None) 917 ) 918 self.body = parsed.query 919 self.headers["content-length"] = str(len(self.body)) 920 921 # Handle retries for server-side errors. 922 resp, content = _retry_request( 923 http, 924 num_retries, 925 "request", 926 self._sleep, 927 self._rand, 928 str(self.uri), 929 method=str(self.method), 930 body=self.body, 931 headers=self.headers, 932 ) 933 934 for callback in self.response_callbacks: 935 callback(resp) 936 if resp.status >= 300: 937 raise HttpError(resp, content, uri=self.uri) 938 return self.postproc(resp, content) 939 940 @util.positional(2) 941 def add_response_callback(self, cb): 942 """add_response_headers_callback 943 944 Args: 945 cb: Callback to be called on receiving the response headers, of signature: 946 947 def cb(resp): 948 # Where resp is an instance of httplib2.Response 949 """ 950 self.response_callbacks.append(cb) 951 952 @util.positional(1) 953 def next_chunk(self, http=None, num_retries=0): 954 """Execute the next step of a resumable upload. 955 956 Can only be used if the method being executed supports media uploads and 957 the MediaUpload object passed in was flagged as using resumable upload. 958 959 Example: 960 961 media = MediaFileUpload('cow.png', mimetype='image/png', 962 chunksize=1000, resumable=True) 963 request = farm.animals().insert( 964 id='cow', 965 name='cow.png', 966 media_body=media) 967 968 response = None 969 while response is None: 970 status, response = request.next_chunk() 971 if status: 972 print "Upload %d%% complete." % int(status.progress() * 100) 973 974 975 Args: 976 http: httplib2.Http, an http object to be used in place of the 977 one the HttpRequest request object was constructed with. 978 num_retries: Integer, number of times to retry with randomized 979 exponential backoff. If all retries fail, the raised HttpError 980 represents the last request. If zero (default), we attempt the 981 request only once. 982 983 Returns: 984 (status, body): (ResumableMediaStatus, object) 985 The body will be None until the resumable media is fully uploaded. 986 987 Raises: 988 googleapiclient.errors.HttpError if the response was not a 2xx. 989 httplib2.HttpLib2Error if a transport error has occurred. 990 """ 991 if http is None: 992 http = self.http 993 994 if self.resumable.size() is None: 995 size = "*" 996 else: 997 size = str(self.resumable.size()) 998 999 if self.resumable_uri is None: 1000 start_headers = copy.copy(self.headers) 1001 start_headers["X-Upload-Content-Type"] = self.resumable.mimetype() 1002 if size != "*": 1003 start_headers["X-Upload-Content-Length"] = size 1004 start_headers["content-length"] = str(self.body_size) 1005 1006 resp, content = _retry_request( 1007 http, 1008 num_retries, 1009 "resumable URI request", 1010 self._sleep, 1011 self._rand, 1012 self.uri, 1013 method=self.method, 1014 body=self.body, 1015 headers=start_headers, 1016 ) 1017 1018 if resp.status == 200 and "location" in resp: 1019 self.resumable_uri = resp["location"] 1020 else: 1021 raise ResumableUploadError(resp, content) 1022 elif self._in_error_state: 1023 # If we are in an error state then query the server for current state of 1024 # the upload by sending an empty PUT and reading the 'range' header in 1025 # the response. 1026 headers = {"Content-Range": "bytes */%s" % size, "content-length": "0"} 1027 resp, content = http.request(self.resumable_uri, "PUT", headers=headers) 1028 status, body = self._process_response(resp, content) 1029 if body: 1030 # The upload was complete. 1031 return (status, body) 1032 1033 if self.resumable.has_stream(): 1034 data = self.resumable.stream() 1035 if self.resumable.chunksize() == -1: 1036 data.seek(self.resumable_progress) 1037 chunk_end = self.resumable.size() - self.resumable_progress - 1 1038 else: 1039 # Doing chunking with a stream, so wrap a slice of the stream. 1040 data = _StreamSlice( 1041 data, self.resumable_progress, self.resumable.chunksize() 1042 ) 1043 chunk_end = min( 1044 self.resumable_progress + self.resumable.chunksize() - 1, 1045 self.resumable.size() - 1, 1046 ) 1047 else: 1048 data = self.resumable.getbytes( 1049 self.resumable_progress, self.resumable.chunksize() 1050 ) 1051 1052 # A short read implies that we are at EOF, so finish the upload. 1053 if len(data) < self.resumable.chunksize(): 1054 size = str(self.resumable_progress + len(data)) 1055 1056 chunk_end = self.resumable_progress + len(data) - 1 1057 1058 headers = { 1059 # Must set the content-length header here because httplib can't 1060 # calculate the size when working with _StreamSlice. 1061 "Content-Length": str(chunk_end - self.resumable_progress + 1), 1062 } 1063 1064 # An empty file results in chunk_end = -1 and size = 0 1065 # sending "bytes 0--1/0" results in an invalid request 1066 # Only add header "Content-Range" if chunk_end != -1 1067 if chunk_end != -1: 1068 headers["Content-Range"] = "bytes %d-%d/%s" % ( 1069 self.resumable_progress, 1070 chunk_end, 1071 size, 1072 ) 1073 1074 for retry_num in range(num_retries + 1): 1075 if retry_num > 0: 1076 self._sleep(self._rand() * 2 ** retry_num) 1077 LOGGER.warning( 1078 "Retry #%d for media upload: %s %s, following status: %d" 1079 % (retry_num, self.method, self.uri, resp.status) 1080 ) 1081 1082 try: 1083 resp, content = http.request( 1084 self.resumable_uri, method="PUT", body=data, headers=headers 1085 ) 1086 except: 1087 self._in_error_state = True 1088 raise 1089 if not _should_retry_response(resp.status, content): 1090 break 1091 1092 return self._process_response(resp, content) 1093 1094 def _process_response(self, resp, content): 1095 """Process the response from a single chunk upload. 1096 1097 Args: 1098 resp: httplib2.Response, the response object. 1099 content: string, the content of the response. 1100 1101 Returns: 1102 (status, body): (ResumableMediaStatus, object) 1103 The body will be None until the resumable media is fully uploaded. 1104 1105 Raises: 1106 googleapiclient.errors.HttpError if the response was not a 2xx or a 308. 1107 """ 1108 if resp.status in [200, 201]: 1109 self._in_error_state = False 1110 return None, self.postproc(resp, content) 1111 elif resp.status == 308: 1112 self._in_error_state = False 1113 # A "308 Resume Incomplete" indicates we are not done. 1114 try: 1115 self.resumable_progress = int(resp["range"].split("-")[1]) + 1 1116 except KeyError: 1117 # If resp doesn't contain range header, resumable progress is 0 1118 self.resumable_progress = 0 1119 if "location" in resp: 1120 self.resumable_uri = resp["location"] 1121 else: 1122 self._in_error_state = True 1123 raise HttpError(resp, content, uri=self.uri) 1124 1125 return ( 1126 MediaUploadProgress(self.resumable_progress, self.resumable.size()), 1127 None, 1128 ) 1129 1130 def to_json(self): 1131 """Returns a JSON representation of the HttpRequest.""" 1132 d = copy.copy(self.__dict__) 1133 if d["resumable"] is not None: 1134 d["resumable"] = self.resumable.to_json() 1135 del d["http"] 1136 del d["postproc"] 1137 del d["_sleep"] 1138 del d["_rand"] 1139 1140 return json.dumps(d) 1141 1142 @staticmethod 1143 def from_json(s, http, postproc): 1144 """Returns an HttpRequest populated with info from a JSON object.""" 1145 d = json.loads(s) 1146 if d["resumable"] is not None: 1147 d["resumable"] = MediaUpload.new_from_json(d["resumable"]) 1148 return HttpRequest( 1149 http, 1150 postproc, 1151 uri=d["uri"], 1152 method=d["method"], 1153 body=d["body"], 1154 headers=d["headers"], 1155 methodId=d["methodId"], 1156 resumable=d["resumable"], 1157 ) 1158 1159 @staticmethod 1160 def null_postproc(resp, contents): 1161 return resp, contents 1162 1163 1164class BatchHttpRequest(object): 1165 """Batches multiple HttpRequest objects into a single HTTP request. 1166 1167 Example: 1168 from googleapiclient.http import BatchHttpRequest 1169 1170 def list_animals(request_id, response, exception): 1171 \"\"\"Do something with the animals list response.\"\"\" 1172 if exception is not None: 1173 # Do something with the exception. 1174 pass 1175 else: 1176 # Do something with the response. 1177 pass 1178 1179 def list_farmers(request_id, response, exception): 1180 \"\"\"Do something with the farmers list response.\"\"\" 1181 if exception is not None: 1182 # Do something with the exception. 1183 pass 1184 else: 1185 # Do something with the response. 1186 pass 1187 1188 service = build('farm', 'v2') 1189 1190 batch = BatchHttpRequest() 1191 1192 batch.add(service.animals().list(), list_animals) 1193 batch.add(service.farmers().list(), list_farmers) 1194 batch.execute(http=http) 1195 """ 1196 1197 @util.positional(1) 1198 def __init__(self, callback=None, batch_uri=None): 1199 """Constructor for a BatchHttpRequest. 1200 1201 Args: 1202 callback: callable, A callback to be called for each response, of the 1203 form callback(id, response, exception). The first parameter is the 1204 request id, and the second is the deserialized response object. The 1205 third is an googleapiclient.errors.HttpError exception object if an HTTP error 1206 occurred while processing the request, or None if no error occurred. 1207 batch_uri: string, URI to send batch requests to. 1208 """ 1209 if batch_uri is None: 1210 batch_uri = _LEGACY_BATCH_URI 1211 1212 if batch_uri == _LEGACY_BATCH_URI: 1213 LOGGER.warning( 1214 "You have constructed a BatchHttpRequest using the legacy batch " 1215 "endpoint %s. This endpoint will be turned down on August 12, 2020. " 1216 "Please provide the API-specific endpoint or use " 1217 "service.new_batch_http_request(). For more details see " 1218 "https://developers.googleblog.com/2018/03/discontinuing-support-for-json-rpc-and.html" 1219 "and https://developers.google.com/api-client-library/python/guide/batch.", 1220 _LEGACY_BATCH_URI, 1221 ) 1222 self._batch_uri = batch_uri 1223 1224 # Global callback to be called for each individual response in the batch. 1225 self._callback = callback 1226 1227 # A map from id to request. 1228 self._requests = {} 1229 1230 # A map from id to callback. 1231 self._callbacks = {} 1232 1233 # List of request ids, in the order in which they were added. 1234 self._order = [] 1235 1236 # The last auto generated id. 1237 self._last_auto_id = 0 1238 1239 # Unique ID on which to base the Content-ID headers. 1240 self._base_id = None 1241 1242 # A map from request id to (httplib2.Response, content) response pairs 1243 self._responses = {} 1244 1245 # A map of id(Credentials) that have been refreshed. 1246 self._refreshed_credentials = {} 1247 1248 def _refresh_and_apply_credentials(self, request, http): 1249 """Refresh the credentials and apply to the request. 1250 1251 Args: 1252 request: HttpRequest, the request. 1253 http: httplib2.Http, the global http object for the batch. 1254 """ 1255 # For the credentials to refresh, but only once per refresh_token 1256 # If there is no http per the request then refresh the http passed in 1257 # via execute() 1258 creds = None 1259 request_credentials = False 1260 1261 if request.http is not None: 1262 creds = _auth.get_credentials_from_http(request.http) 1263 request_credentials = True 1264 1265 if creds is None and http is not None: 1266 creds = _auth.get_credentials_from_http(http) 1267 1268 if creds is not None: 1269 if id(creds) not in self._refreshed_credentials: 1270 _auth.refresh_credentials(creds) 1271 self._refreshed_credentials[id(creds)] = 1 1272 1273 # Only apply the credentials if we are using the http object passed in, 1274 # otherwise apply() will get called during _serialize_request(). 1275 if request.http is None or not request_credentials: 1276 _auth.apply_credentials(creds, request.headers) 1277 1278 def _id_to_header(self, id_): 1279 """Convert an id to a Content-ID header value. 1280 1281 Args: 1282 id_: string, identifier of individual request. 1283 1284 Returns: 1285 A Content-ID header with the id_ encoded into it. A UUID is prepended to 1286 the value because Content-ID headers are supposed to be universally 1287 unique. 1288 """ 1289 if self._base_id is None: 1290 self._base_id = uuid.uuid4() 1291 1292 # NB: we intentionally leave whitespace between base/id and '+', so RFC2822 1293 # line folding works properly on Python 3; see 1294 # https://github.com/googleapis/google-api-python-client/issues/164 1295 return "<%s + %s>" % (self._base_id, urllib.parse.quote(id_)) 1296 1297 def _header_to_id(self, header): 1298 """Convert a Content-ID header value to an id. 1299 1300 Presumes the Content-ID header conforms to the format that _id_to_header() 1301 returns. 1302 1303 Args: 1304 header: string, Content-ID header value. 1305 1306 Returns: 1307 The extracted id value. 1308 1309 Raises: 1310 BatchError if the header is not in the expected format. 1311 """ 1312 if header[0] != "<" or header[-1] != ">": 1313 raise BatchError("Invalid value for Content-ID: %s" % header) 1314 if "+" not in header: 1315 raise BatchError("Invalid value for Content-ID: %s" % header) 1316 base, id_ = header[1:-1].split(" + ", 1) 1317 1318 return urllib.parse.unquote(id_) 1319 1320 def _serialize_request(self, request): 1321 """Convert an HttpRequest object into a string. 1322 1323 Args: 1324 request: HttpRequest, the request to serialize. 1325 1326 Returns: 1327 The request as a string in application/http format. 1328 """ 1329 # Construct status line 1330 parsed = urllib.parse.urlparse(request.uri) 1331 request_line = urllib.parse.urlunparse( 1332 ("", "", parsed.path, parsed.params, parsed.query, "") 1333 ) 1334 status_line = request.method + " " + request_line + " HTTP/1.1\n" 1335 major, minor = request.headers.get("content-type", "application/json").split( 1336 "/" 1337 ) 1338 msg = MIMENonMultipart(major, minor) 1339 headers = request.headers.copy() 1340 1341 if request.http is not None: 1342 credentials = _auth.get_credentials_from_http(request.http) 1343 if credentials is not None: 1344 _auth.apply_credentials(credentials, headers) 1345 1346 # MIMENonMultipart adds its own Content-Type header. 1347 if "content-type" in headers: 1348 del headers["content-type"] 1349 1350 for key, value in headers.items(): 1351 msg[key] = value 1352 msg["Host"] = parsed.netloc 1353 msg.set_unixfrom(None) 1354 1355 if request.body is not None: 1356 msg.set_payload(request.body) 1357 msg["content-length"] = str(len(request.body)) 1358 1359 # Serialize the mime message. 1360 fp = io.StringIO() 1361 # maxheaderlen=0 means don't line wrap headers. 1362 g = Generator(fp, maxheaderlen=0) 1363 g.flatten(msg, unixfrom=False) 1364 body = fp.getvalue() 1365 1366 return status_line + body 1367 1368 def _deserialize_response(self, payload): 1369 """Convert string into httplib2 response and content. 1370 1371 Args: 1372 payload: string, headers and body as a string. 1373 1374 Returns: 1375 A pair (resp, content), such as would be returned from httplib2.request. 1376 """ 1377 # Strip off the status line 1378 status_line, payload = payload.split("\n", 1) 1379 protocol, status, reason = status_line.split(" ", 2) 1380 1381 # Parse the rest of the response 1382 parser = FeedParser() 1383 parser.feed(payload) 1384 msg = parser.close() 1385 msg["status"] = status 1386 1387 # Create httplib2.Response from the parsed headers. 1388 resp = httplib2.Response(msg) 1389 resp.reason = reason 1390 resp.version = int(protocol.split("/", 1)[1].replace(".", "")) 1391 1392 content = payload.split("\r\n\r\n", 1)[1] 1393 1394 return resp, content 1395 1396 def _new_id(self): 1397 """Create a new id. 1398 1399 Auto incrementing number that avoids conflicts with ids already used. 1400 1401 Returns: 1402 string, a new unique id. 1403 """ 1404 self._last_auto_id += 1 1405 while str(self._last_auto_id) in self._requests: 1406 self._last_auto_id += 1 1407 return str(self._last_auto_id) 1408 1409 @util.positional(2) 1410 def add(self, request, callback=None, request_id=None): 1411 """Add a new request. 1412 1413 Every callback added will be paired with a unique id, the request_id. That 1414 unique id will be passed back to the callback when the response comes back 1415 from the server. The default behavior is to have the library generate it's 1416 own unique id. If the caller passes in a request_id then they must ensure 1417 uniqueness for each request_id, and if they are not an exception is 1418 raised. Callers should either supply all request_ids or never supply a 1419 request id, to avoid such an error. 1420 1421 Args: 1422 request: HttpRequest, Request to add to the batch. 1423 callback: callable, A callback to be called for this response, of the 1424 form callback(id, response, exception). The first parameter is the 1425 request id, and the second is the deserialized response object. The 1426 third is an googleapiclient.errors.HttpError exception object if an HTTP error 1427 occurred while processing the request, or None if no errors occurred. 1428 request_id: string, A unique id for the request. The id will be passed 1429 to the callback with the response. 1430 1431 Returns: 1432 None 1433 1434 Raises: 1435 BatchError if a media request is added to a batch. 1436 KeyError is the request_id is not unique. 1437 """ 1438 1439 if len(self._order) >= MAX_BATCH_LIMIT: 1440 raise BatchError( 1441 "Exceeded the maximum calls(%d) in a single batch request." 1442 % MAX_BATCH_LIMIT 1443 ) 1444 if request_id is None: 1445 request_id = self._new_id() 1446 if request.resumable is not None: 1447 raise BatchError("Media requests cannot be used in a batch request.") 1448 if request_id in self._requests: 1449 raise KeyError("A request with this ID already exists: %s" % request_id) 1450 self._requests[request_id] = request 1451 self._callbacks[request_id] = callback 1452 self._order.append(request_id) 1453 1454 def _execute(self, http, order, requests): 1455 """Serialize batch request, send to server, process response. 1456 1457 Args: 1458 http: httplib2.Http, an http object to be used to make the request with. 1459 order: list, list of request ids in the order they were added to the 1460 batch. 1461 requests: list, list of request objects to send. 1462 1463 Raises: 1464 httplib2.HttpLib2Error if a transport error has occurred. 1465 googleapiclient.errors.BatchError if the response is the wrong format. 1466 """ 1467 message = MIMEMultipart("mixed") 1468 # Message should not write out it's own headers. 1469 setattr(message, "_write_headers", lambda self: None) 1470 1471 # Add all the individual requests. 1472 for request_id in order: 1473 request = requests[request_id] 1474 1475 msg = MIMENonMultipart("application", "http") 1476 msg["Content-Transfer-Encoding"] = "binary" 1477 msg["Content-ID"] = self._id_to_header(request_id) 1478 1479 body = self._serialize_request(request) 1480 msg.set_payload(body) 1481 message.attach(msg) 1482 1483 # encode the body: note that we can't use `as_string`, because 1484 # it plays games with `From ` lines. 1485 fp = io.StringIO() 1486 g = Generator(fp, mangle_from_=False) 1487 g.flatten(message, unixfrom=False) 1488 body = fp.getvalue() 1489 1490 headers = {} 1491 headers["content-type"] = ( 1492 "multipart/mixed; " 'boundary="%s"' 1493 ) % message.get_boundary() 1494 1495 resp, content = http.request( 1496 self._batch_uri, method="POST", body=body, headers=headers 1497 ) 1498 1499 if resp.status >= 300: 1500 raise HttpError(resp, content, uri=self._batch_uri) 1501 1502 # Prepend with a content-type header so FeedParser can handle it. 1503 header = "content-type: %s\r\n\r\n" % resp["content-type"] 1504 # PY3's FeedParser only accepts unicode. So we should decode content 1505 # here, and encode each payload again. 1506 content = content.decode("utf-8") 1507 for_parser = header + content 1508 1509 parser = FeedParser() 1510 parser.feed(for_parser) 1511 mime_response = parser.close() 1512 1513 if not mime_response.is_multipart(): 1514 raise BatchError( 1515 "Response not in multipart/mixed format.", resp=resp, content=content 1516 ) 1517 1518 for part in mime_response.get_payload(): 1519 request_id = self._header_to_id(part["Content-ID"]) 1520 response, content = self._deserialize_response(part.get_payload()) 1521 # We encode content here to emulate normal http response. 1522 if isinstance(content, str): 1523 content = content.encode("utf-8") 1524 self._responses[request_id] = (response, content) 1525 1526 @util.positional(1) 1527 def execute(self, http=None): 1528 """Execute all the requests as a single batched HTTP request. 1529 1530 Args: 1531 http: httplib2.Http, an http object to be used in place of the one the 1532 HttpRequest request object was constructed with. If one isn't supplied 1533 then use a http object from the requests in this batch. 1534 1535 Returns: 1536 None 1537 1538 Raises: 1539 httplib2.HttpLib2Error if a transport error has occurred. 1540 googleapiclient.errors.BatchError if the response is the wrong format. 1541 """ 1542 # If we have no requests return 1543 if len(self._order) == 0: 1544 return None 1545 1546 # If http is not supplied use the first valid one given in the requests. 1547 if http is None: 1548 for request_id in self._order: 1549 request = self._requests[request_id] 1550 if request is not None: 1551 http = request.http 1552 break 1553 1554 if http is None: 1555 raise ValueError("Missing a valid http object.") 1556 1557 # Special case for OAuth2Credentials-style objects which have not yet been 1558 # refreshed with an initial access_token. 1559 creds = _auth.get_credentials_from_http(http) 1560 if creds is not None: 1561 if not _auth.is_valid(creds): 1562 LOGGER.info("Attempting refresh to obtain initial access_token") 1563 _auth.refresh_credentials(creds) 1564 1565 self._execute(http, self._order, self._requests) 1566 1567 # Loop over all the requests and check for 401s. For each 401 request the 1568 # credentials should be refreshed and then sent again in a separate batch. 1569 redo_requests = {} 1570 redo_order = [] 1571 1572 for request_id in self._order: 1573 resp, content = self._responses[request_id] 1574 if resp["status"] == "401": 1575 redo_order.append(request_id) 1576 request = self._requests[request_id] 1577 self._refresh_and_apply_credentials(request, http) 1578 redo_requests[request_id] = request 1579 1580 if redo_requests: 1581 self._execute(http, redo_order, redo_requests) 1582 1583 # Now process all callbacks that are erroring, and raise an exception for 1584 # ones that return a non-2xx response? Or add extra parameter to callback 1585 # that contains an HttpError? 1586 1587 for request_id in self._order: 1588 resp, content = self._responses[request_id] 1589 1590 request = self._requests[request_id] 1591 callback = self._callbacks[request_id] 1592 1593 response = None 1594 exception = None 1595 try: 1596 if resp.status >= 300: 1597 raise HttpError(resp, content, uri=request.uri) 1598 response = request.postproc(resp, content) 1599 except HttpError as e: 1600 exception = e 1601 1602 if callback is not None: 1603 callback(request_id, response, exception) 1604 if self._callback is not None: 1605 self._callback(request_id, response, exception) 1606 1607 1608class HttpRequestMock(object): 1609 """Mock of HttpRequest. 1610 1611 Do not construct directly, instead use RequestMockBuilder. 1612 """ 1613 1614 def __init__(self, resp, content, postproc): 1615 """Constructor for HttpRequestMock 1616 1617 Args: 1618 resp: httplib2.Response, the response to emulate coming from the request 1619 content: string, the response body 1620 postproc: callable, the post processing function usually supplied by 1621 the model class. See model.JsonModel.response() as an example. 1622 """ 1623 self.resp = resp 1624 self.content = content 1625 self.postproc = postproc 1626 if resp is None: 1627 self.resp = httplib2.Response({"status": 200, "reason": "OK"}) 1628 if "reason" in self.resp: 1629 self.resp.reason = self.resp["reason"] 1630 1631 def execute(self, http=None): 1632 """Execute the request. 1633 1634 Same behavior as HttpRequest.execute(), but the response is 1635 mocked and not really from an HTTP request/response. 1636 """ 1637 return self.postproc(self.resp, self.content) 1638 1639 1640class RequestMockBuilder(object): 1641 """A simple mock of HttpRequest 1642 1643 Pass in a dictionary to the constructor that maps request methodIds to 1644 tuples of (httplib2.Response, content, opt_expected_body) that should be 1645 returned when that method is called. None may also be passed in for the 1646 httplib2.Response, in which case a 200 OK response will be generated. 1647 If an opt_expected_body (str or dict) is provided, it will be compared to 1648 the body and UnexpectedBodyError will be raised on inequality. 1649 1650 Example: 1651 response = '{"data": {"id": "tag:google.c...' 1652 requestBuilder = RequestMockBuilder( 1653 { 1654 'plus.activities.get': (None, response), 1655 } 1656 ) 1657 googleapiclient.discovery.build("plus", "v1", requestBuilder=requestBuilder) 1658 1659 Methods that you do not supply a response for will return a 1660 200 OK with an empty string as the response content or raise an excpetion 1661 if check_unexpected is set to True. The methodId is taken from the rpcName 1662 in the discovery document. 1663 1664 For more details see the project wiki. 1665 """ 1666 1667 def __init__(self, responses, check_unexpected=False): 1668 """Constructor for RequestMockBuilder 1669 1670 The constructed object should be a callable object 1671 that can replace the class HttpResponse. 1672 1673 responses - A dictionary that maps methodIds into tuples 1674 of (httplib2.Response, content). The methodId 1675 comes from the 'rpcName' field in the discovery 1676 document. 1677 check_unexpected - A boolean setting whether or not UnexpectedMethodError 1678 should be raised on unsupplied method. 1679 """ 1680 self.responses = responses 1681 self.check_unexpected = check_unexpected 1682 1683 def __call__( 1684 self, 1685 http, 1686 postproc, 1687 uri, 1688 method="GET", 1689 body=None, 1690 headers=None, 1691 methodId=None, 1692 resumable=None, 1693 ): 1694 """Implements the callable interface that discovery.build() expects 1695 of requestBuilder, which is to build an object compatible with 1696 HttpRequest.execute(). See that method for the description of the 1697 parameters and the expected response. 1698 """ 1699 if methodId in self.responses: 1700 response = self.responses[methodId] 1701 resp, content = response[:2] 1702 if len(response) > 2: 1703 # Test the body against the supplied expected_body. 1704 expected_body = response[2] 1705 if bool(expected_body) != bool(body): 1706 # Not expecting a body and provided one 1707 # or expecting a body and not provided one. 1708 raise UnexpectedBodyError(expected_body, body) 1709 if isinstance(expected_body, str): 1710 expected_body = json.loads(expected_body) 1711 body = json.loads(body) 1712 if body != expected_body: 1713 raise UnexpectedBodyError(expected_body, body) 1714 return HttpRequestMock(resp, content, postproc) 1715 elif self.check_unexpected: 1716 raise UnexpectedMethodError(methodId=methodId) 1717 else: 1718 model = JsonModel(False) 1719 return HttpRequestMock(None, "{}", model.response) 1720 1721 1722class HttpMock(object): 1723 """Mock of httplib2.Http""" 1724 1725 def __init__(self, filename=None, headers=None): 1726 """ 1727 Args: 1728 filename: string, absolute filename to read response from 1729 headers: dict, header to return with response 1730 """ 1731 if headers is None: 1732 headers = {"status": "200"} 1733 if filename: 1734 with open(filename, "rb") as f: 1735 self.data = f.read() 1736 else: 1737 self.data = None 1738 self.response_headers = headers 1739 self.headers = None 1740 self.uri = None 1741 self.method = None 1742 self.body = None 1743 self.headers = None 1744 1745 def request( 1746 self, 1747 uri, 1748 method="GET", 1749 body=None, 1750 headers=None, 1751 redirections=1, 1752 connection_type=None, 1753 ): 1754 self.uri = uri 1755 self.method = method 1756 self.body = body 1757 self.headers = headers 1758 return httplib2.Response(self.response_headers), self.data 1759 1760 def close(self): 1761 return None 1762 1763 1764class HttpMockSequence(object): 1765 """Mock of httplib2.Http 1766 1767 Mocks a sequence of calls to request returning different responses for each 1768 call. Create an instance initialized with the desired response headers 1769 and content and then use as if an httplib2.Http instance. 1770 1771 http = HttpMockSequence([ 1772 ({'status': '401'}, ''), 1773 ({'status': '200'}, '{"access_token":"1/3w","expires_in":3600}'), 1774 ({'status': '200'}, 'echo_request_headers'), 1775 ]) 1776 resp, content = http.request("http://examples.com") 1777 1778 There are special values you can pass in for content to trigger 1779 behavours that are helpful in testing. 1780 1781 'echo_request_headers' means return the request headers in the response body 1782 'echo_request_headers_as_json' means return the request headers in 1783 the response body 1784 'echo_request_body' means return the request body in the response body 1785 'echo_request_uri' means return the request uri in the response body 1786 """ 1787 1788 def __init__(self, iterable): 1789 """ 1790 Args: 1791 iterable: iterable, a sequence of pairs of (headers, body) 1792 """ 1793 self._iterable = iterable 1794 self.follow_redirects = True 1795 self.request_sequence = list() 1796 1797 def request( 1798 self, 1799 uri, 1800 method="GET", 1801 body=None, 1802 headers=None, 1803 redirections=1, 1804 connection_type=None, 1805 ): 1806 # Remember the request so after the fact this mock can be examined 1807 self.request_sequence.append((uri, method, body, headers)) 1808 resp, content = self._iterable.pop(0) 1809 if isinstance(content, str): 1810 content = content.encode("utf-8") 1811 1812 if content == b"echo_request_headers": 1813 content = headers 1814 elif content == b"echo_request_headers_as_json": 1815 content = json.dumps(headers) 1816 elif content == b"echo_request_body": 1817 if hasattr(body, "read"): 1818 content = body.read() 1819 else: 1820 content = body 1821 elif content == b"echo_request_uri": 1822 content = uri 1823 if isinstance(content, str): 1824 content = content.encode("utf-8") 1825 return httplib2.Response(resp), content 1826 1827 1828def set_user_agent(http, user_agent): 1829 """Set the user-agent on every request. 1830 1831 Args: 1832 http - An instance of httplib2.Http 1833 or something that acts like it. 1834 user_agent: string, the value for the user-agent header. 1835 1836 Returns: 1837 A modified instance of http that was passed in. 1838 1839 Example: 1840 1841 h = httplib2.Http() 1842 h = set_user_agent(h, "my-app-name/6.0") 1843 1844 Most of the time the user-agent will be set doing auth, this is for the rare 1845 cases where you are accessing an unauthenticated endpoint. 1846 """ 1847 request_orig = http.request 1848 1849 # The closure that will replace 'httplib2.Http.request'. 1850 def new_request( 1851 uri, 1852 method="GET", 1853 body=None, 1854 headers=None, 1855 redirections=httplib2.DEFAULT_MAX_REDIRECTS, 1856 connection_type=None, 1857 ): 1858 """Modify the request headers to add the user-agent.""" 1859 if headers is None: 1860 headers = {} 1861 if "user-agent" in headers: 1862 headers["user-agent"] = user_agent + " " + headers["user-agent"] 1863 else: 1864 headers["user-agent"] = user_agent 1865 resp, content = request_orig( 1866 uri, 1867 method=method, 1868 body=body, 1869 headers=headers, 1870 redirections=redirections, 1871 connection_type=connection_type, 1872 ) 1873 return resp, content 1874 1875 http.request = new_request 1876 return http 1877 1878 1879def tunnel_patch(http): 1880 """Tunnel PATCH requests over POST. 1881 Args: 1882 http - An instance of httplib2.Http 1883 or something that acts like it. 1884 1885 Returns: 1886 A modified instance of http that was passed in. 1887 1888 Example: 1889 1890 h = httplib2.Http() 1891 h = tunnel_patch(h, "my-app-name/6.0") 1892 1893 Useful if you are running on a platform that doesn't support PATCH. 1894 Apply this last if you are using OAuth 1.0, as changing the method 1895 will result in a different signature. 1896 """ 1897 request_orig = http.request 1898 1899 # The closure that will replace 'httplib2.Http.request'. 1900 def new_request( 1901 uri, 1902 method="GET", 1903 body=None, 1904 headers=None, 1905 redirections=httplib2.DEFAULT_MAX_REDIRECTS, 1906 connection_type=None, 1907 ): 1908 """Modify the request headers to add the user-agent.""" 1909 if headers is None: 1910 headers = {} 1911 if method == "PATCH": 1912 if "oauth_token" in headers.get("authorization", ""): 1913 LOGGER.warning( 1914 "OAuth 1.0 request made with Credentials after tunnel_patch." 1915 ) 1916 headers["x-http-method-override"] = "PATCH" 1917 method = "POST" 1918 resp, content = request_orig( 1919 uri, 1920 method=method, 1921 body=body, 1922 headers=headers, 1923 redirections=redirections, 1924 connection_type=connection_type, 1925 ) 1926 return resp, content 1927 1928 http.request = new_request 1929 return http 1930 1931 1932def build_http(): 1933 """Builds httplib2.Http object 1934 1935 Returns: 1936 A httplib2.Http object, which is used to make http requests, and which has timeout set by default. 1937 To override default timeout call 1938 1939 socket.setdefaulttimeout(timeout_in_sec) 1940 1941 before interacting with this method. 1942 """ 1943 if socket.getdefaulttimeout() is not None: 1944 http_timeout = socket.getdefaulttimeout() 1945 else: 1946 http_timeout = DEFAULT_HTTP_TIMEOUT_SEC 1947 http = httplib2.Http(timeout=http_timeout) 1948 # 308's are used by several Google APIs (Drive, YouTube) 1949 # for Resumable Uploads rather than Permanent Redirects. 1950 # This asks httplib2 to exclude 308s from the status codes 1951 # it treats as redirects 1952 try: 1953 http.redirect_codes = http.redirect_codes - {308} 1954 except AttributeError: 1955 # Apache Beam tests depend on this library and cannot 1956 # currently upgrade their httplib2 version 1957 # http.redirect_codes does not exist in previous versions 1958 # of httplib2, so pass 1959 pass 1960 1961 return http 1962