tessl/pypi-google-api-python-client
Google API Client Library for Python that provides discovery-based access to hundreds of Google services with authentication, caching, and media upload/download support.
—
Pending
http.mddocs/
HTTP Request Handling
The http module provides comprehensive HTTP functionality including individual request execution, batch processing, retry logic, authentication integration, and testing utilities.
Capabilities
Individual HTTP Requests
Execute individual HTTP requests with authentication, retry logic, and response processing.
class HttpRequest:
"""Represents an HTTP request to be executed."""
def __init__(self, http, postproc, uri, method='GET', body=None,
headers=None, methodId=None, resumable=None):
"""
Initialize an HTTP request.
Args:
http (httplib2.Http): HTTP client to use for the request
postproc (callable): Function to process the response
uri (str): URI for the HTTP request
method (str): HTTP method ('GET', 'POST', 'PUT', 'DELETE', etc.)
body (str, optional): Request body content
headers (dict, optional): HTTP headers for the request
methodId (str, optional): API method identifier for logging
resumable (MediaUpload, optional): Resumable media upload
"""
def execute(self, http=None, num_retries=0):
"""
Execute the HTTP request.
Args:
http (httplib2.Http, optional): HTTP client to use (overrides default)
num_retries (int): Number of times to retry on recoverable errors
Returns:
object: Deserialized response content based on the response model
Raises:
HttpError: When the HTTP request fails with an error status
ResumableUploadError: When resumable upload fails
"""
def add_response_callback(self, cb):
"""
Add a callback function to process the HTTP response.
Args:
cb (callable): Callback function that takes (resp, content) arguments
"""
def to_json(self):
"""
Serialize the request to JSON for storage or transmission.
Returns:
str: JSON representation of the request
"""
@staticmethod
def from_json(s, http, postproc):
"""
Deserialize a request from JSON.
Args:
s (str): JSON string representing a request
http (httplib2.Http): HTTP client instance
postproc (callable): Response post-processing function
Returns:
HttpRequest: Reconstructed request object
"""
def next_chunk(self, http=None, num_retries=0):
"""
Download the next chunk of a resumable media upload.
Args:
http (httplib2.Http, optional): HTTP client to use
num_retries (int): Number of retry attempts
Returns:
tuple: (MediaUploadProgress or None, object or None) -
Progress status and response body (None until upload complete)
"""
@staticmethod
def null_postproc(resp, contents):
"""
Default post-processing function that returns response as-is.
Args:
resp: HTTP response object
contents: Response body content
Returns:
tuple: (resp, contents) unchanged
"""Batch HTTP Requests
Execute multiple HTTP requests efficiently in a single batch operation.
class BatchHttpRequest:
"""Batch multiple HTTP requests for efficient execution."""
def __init__(self, callback=None, batch_uri=None):
"""
Initialize a batch HTTP request.
Args:
callback (callable, optional): Default callback for all requests
batch_uri (str, optional): Custom URI for batch endpoint
"""
def add(self, request, callback=None, request_id=None):
"""
Add a request to the batch.
Args:
request (HttpRequest): The request to add to the batch
callback (callable, optional): Callback for this specific request
request_id (str, optional): Unique identifier for this request
Raises:
BatchError: When the batch is full or request is invalid
"""
def execute(self, http=None, num_retries=0):
"""
Execute all requests in the batch.
Args:
http (httplib2.Http, optional): HTTP client to use
num_retries (int): Number of retry attempts for the batch
Raises:
BatchError: When batch execution fails
"""HTTP Constants
DEFAULT_CHUNK_SIZE = 100 * 1024 * 1024 # 100MB default chunk size
MAX_URI_LENGTH = 2048 # Maximum URI length for GET requests
MAX_BATCH_LIMIT = 1000 # Maximum requests per batch
DEFAULT_HTTP_TIMEOUT_SEC = 60 # Default HTTP timeout in secondsMedia Upload Progress
Progress tracking for resumable media uploads.
class MediaUploadProgress:
"""Status of a resumable upload."""
def __init__(self, resumable_progress, total_size):
"""
Initialize upload progress.
Args:
resumable_progress (int): Bytes sent so far
total_size (int or None): Total bytes in complete upload, or None if unknown
"""
def progress(self):
"""
Get upload progress as a percentage.
Returns:
float: Percent of upload completed (0.0 if total size unknown)
"""Media Download Progress
Progress tracking for media downloads.
class MediaDownloadProgress:
"""Status of a resumable download."""
def __init__(self, resumable_progress, total_size):
"""
Initialize download progress.
Args:
resumable_progress (int): Bytes received so far
total_size (int): Total bytes in complete download
"""
def progress(self):
"""
Get download progress as a percentage.
Returns:
float: Percent of download completed (0.0 if total size unknown)
"""HTTP Testing Utilities
Mock classes for testing HTTP requests and responses.
class HttpMock:
"""Mock of httplib2.Http for testing."""
def __init__(self, filename=None, headers=None):
"""
Initialize HTTP mock.
Args:
filename (str, optional): Absolute path to response file
headers (dict, optional): Headers to return (default: {"status": "200"})
"""
def request(self, uri, method="GET", body=None, headers=None,
redirections=1, connection_type=None):
"""
Mock HTTP request.
Args:
uri (str): Request URI
method (str): HTTP method
body (str, optional): Request body
headers (dict, optional): Request headers
redirections (int): Number of redirections to follow
connection_type: Connection type
Returns:
tuple: (httplib2.Response, bytes or None) - mock response
"""
def close(self):
"""Close connections (compatibility method)."""
class HttpMockSequence:
"""Mock a sequence of HTTP responses for testing."""
def __init__(self, iterable):
"""
Initialize mock sequence.
Args:
iterable: Sequence of (headers, body) pairs for sequential responses
"""
def request(self, uri, method="GET", body=None, headers=None,
redirections=1, connection_type=None):
"""
Return sequential mock responses.
Returns:
tuple: (httplib2.Response, bytes) - next response in sequence
"""HTTP Utility Functions
Utility functions for HTTP client configuration and processing.
def set_user_agent(http, user_agent):
"""
Set the user-agent header on every request.
Args:
http (httplib2.Http): HTTP client instance
user_agent (str): User-agent string
Returns:
httplib2.Http: Modified HTTP client with user-agent injection
"""
def tunnel_patch(http):
"""
Tunnel PATCH requests over POST for OAuth 1.0 compatibility.
Args:
http (httplib2.Http): HTTP client instance
Returns:
httplib2.Http: Modified HTTP client with PATCH tunneling
"""
def build_http():
"""
Build an httplib2.Http object with default timeout and redirect handling.
Returns:
httplib2.Http: HTTP client with default configuration
"""Usage Examples
Basic Request Execution
from googleapiclient import discovery
# Build service and create request
service = discovery.build('gmail', 'v1', credentials=credentials)
request = service.users().messages().list(userId='me', maxResults=10)
# Execute with retry
try:
response = request.execute(num_retries=3)
print(f"Found {len(response.get('messages', []))} messages")
except Exception as e:
print(f"Request failed: {e}")Request Serialization and Deserialization
from googleapiclient import discovery
from googleapiclient.http import HttpRequest
import json
# Create and serialize request
service = discovery.build('gmail', 'v1', credentials=credentials)
request = service.users().messages().get(userId='me', id='message_id')
request_json = request.to_json()
# Store or transmit request_json...
# Recreate request from JSON
restored_request = HttpRequest.from_json(request_json)
response = restored_request.execute()Response Callbacks
def response_callback(resp, content):
print(f"Response status: {resp.status}")
print(f"Response headers: {resp}")
print(f"Content length: {len(content) if content else 0}")
request = service.users().messages().list(userId='me')
request.add_response_callback(response_callback)
response = request.execute()Batch Request Processing
from googleapiclient import http
def batch_callback(request_id, response, exception):
"""Handle individual batch request results."""
if exception is not None:
print(f'Request {request_id} failed: {exception}')
else:
print(f'Request {request_id} succeeded: {response}')
# Create batch request
batch = http.BatchHttpRequest(callback=batch_callback)
# Add multiple requests to batch
message_ids = ['msg1', 'msg2', 'msg3', 'msg4', 'msg5']
for i, msg_id in enumerate(message_ids):
request = service.users().messages().get(userId='me', id=msg_id)
batch.add(request, request_id=f'message_{i}')
# Execute all requests in batch
batch.execute()Batch with Custom Callbacks
def process_message(request_id, response, exception):
if exception:
print(f"Failed to get message {request_id}: {exception}")
else:
subject = 'No subject'
for header in response.get('payload', {}).get('headers', []):
if header['name'] == 'Subject':
subject = header['value']
break
print(f"Message {request_id}: {subject}")
def process_label(request_id, response, exception):
if exception:
print(f"Failed to get label {request_id}: {exception}")
else:
print(f"Label {request_id}: {response.get('name', 'Unknown')}")
batch = http.BatchHttpRequest()
# Add requests with different callbacks
batch.add(
service.users().messages().get(userId='me', id='msg1'),
callback=process_message,
request_id='message_1'
)
batch.add(
service.users().labels().get(userId='me', id='INBOX'),
callback=process_label,
request_id='inbox_label'
)
batch.execute()Error Handling with Retries
from googleapiclient.errors import HttpError
import time
def execute_with_backoff(request, max_retries=3):
"""Execute request with exponential backoff."""
for attempt in range(max_retries):
try:
return request.execute()
except HttpError as error:
if error.resp.status in [429, 500, 502, 503, 504]:
if attempt < max_retries - 1:
wait_time = (2 ** attempt) + (random.randint(0, 1000) / 1000)
time.sleep(wait_time)
continue
raise
# Use custom retry logic
request = service.users().messages().list(userId='me')
response = execute_with_backoff(request)Working with Large Result Sets
def get_all_messages(service, user_id='me'):
"""Get all messages using pagination."""
messages = []
request = service.users().messages().list(userId=user_id)
while request is not None:
response = request.execute()
messages.extend(response.get('messages', []))
request = service.users().messages().list_next(request, response)
return messages
all_messages = get_all_messages(service)
print(f"Total messages: {len(all_messages)}")Install with Tessl CLI
npx tessl i tessl/pypi-google-api-python-client@2.181.1