or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

bidirectional-streaming.mdclient-config.mddatetime.mdexceptions.mdgapic-framework.mdiam-policies.mdindex.mdoperations.mdpage-iteration.mdpath-templates.mdprotobuf-helpers.mdretry.mdtimeout.mdtransport.mduniverse-domain.md
tile.json

tessl/pypi-google-api-core

Google API client core library providing common helpers, utilities, and components for Python client libraries

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/google-api-core@2.25.x

To install, run

npx @tessl/cli install tessl/pypi-google-api-core@2.25.0

index.mddocs/

Google API Core

A comprehensive Python library providing common helpers, utilities, and components used by all Google API client libraries. It serves as the foundational layer for Google's Python client libraries, offering shared functionality for authentication, retry logic, operations management, exception handling, and API communication patterns.

Package Information

  • Package Name: google-api-core
  • Language: Python
  • Installation: pip install google-api-core

Core Imports

import google.api_core

Common import patterns for specific functionality:

from google.api_core import exceptions
from google.api_core import retry
from google.api_core import timeout
from google.api_core import operation
from google.api_core import client_options

Basic Usage

from google.api_core import retry
from google.api_core import exceptions
from google.api_core import timeout
import requests

# Configure retry behavior with exponential backoff
retry_config = retry.Retry(
    initial=1.0,
    maximum=60.0,
    multiplier=2.0,
    predicate=retry.if_exception_type(
        exceptions.InternalServerError,
        exceptions.ServiceUnavailable
    )
)

# Apply retry decorator to a function
@retry_config
def make_api_call():
    response = requests.get("https://api.example.com/data")
    if response.status_code >= 500:
        raise exceptions.InternalServerError("Server error")
    return response.json()

# Use timeout decorators
timeout_config = timeout.TimeToDeadlineTimeout(deadline=30.0)

@timeout_config
def timed_operation():
    # Operation that should complete within deadline
    return make_api_call()

Architecture

Google API Core follows a modular design with distinct functional areas:

  • Exception Hierarchy: Comprehensive error mapping between HTTP/gRPC status codes and Python exceptions
  • Retry Framework: Configurable retry mechanisms with exponential backoff and custom predicates
  • Operation Management: Long-running operation polling and management for both sync and async
  • Page Iteration: Standardized pagination patterns for list operations
  • Protocol Buffer Helpers: Utilities for working with Google's protobuf-based APIs
  • Transport Abstractions: Common patterns for gRPC and REST API communication
  • Authentication Integration: Seamless integration with Google Auth libraries

This architecture ensures consistency across all Google Cloud client libraries while providing maximum flexibility for different API patterns and use cases.

Capabilities

Exception Handling

Comprehensive exception hierarchy mapping HTTP and gRPC status codes to specific Python exceptions, with utilities for error conversion and handling across different transport protocols.

class GoogleAPIError(Exception): ...
class GoogleAPICallError(GoogleAPIError): ...
class RetryError(GoogleAPIError): ...

# HTTP Status Exceptions
class ClientError(GoogleAPICallError): ...
class BadRequest(ClientError): ...
class Unauthorized(ClientError): ...
class Forbidden(ClientError): ...
class NotFound(ClientError): ...

class ServerError(GoogleAPICallError): ...
class InternalServerError(ServerError): ...
class ServiceUnavailable(ServerError): ...

def from_http_status(status_code, message, **kwargs): ...
def from_grpc_error(rpc_exc): ...

Exception Handling

Retry Logic

Configurable retry mechanisms with exponential backoff, custom predicates, and support for both synchronous and asynchronous operations including streaming.

class Retry:
    def __init__(self, predicate=None, initial=1.0, maximum=60.0, multiplier=2.0, deadline=120.0): ...

class AsyncRetry:
    def __init__(self, predicate=None, initial=1.0, maximum=60.0, multiplier=2.0, deadline=120.0): ...

def retry_target(target, predicate, sleep_generator, deadline=None, on_error=None): ...
def if_exception_type(*exception_types): ...
def if_transient_error(exception): ...

Retry Logic

Long-Running Operations

Management of Google API long-running operations with polling, cancellation, and both synchronous and asynchronous support.

class Operation:
    def __init__(self, operation, refresh, cancel, result_type=None, metadata_type=None, retry=None): ...
    def done(self): ...
    def result(self, timeout=None): ...
    def cancel(self): ...

class AsyncOperation:
    def __init__(self, operation, refresh, cancel, result_type=None, metadata_type=None, retry=None): ...
    async def done(self): ...
    async def result(self, timeout=None): ...
    async def cancel(self): ...

Operations

Page Iteration

Standardized pagination patterns for list operations with support for both HTTP/JSON and gRPC APIs, including async variants.

class Iterator:
    def __init__(self, client, item_to_value, page_token=None, max_results=None): ...

class HTTPIterator(Iterator):
    def __init__(self, client, api_request, path, item_to_value, page_token=None, max_results=None): ...

class GRPCIterator(Iterator):
    def __init__(self, client, method, request, items_field, request_token_field="page_token"): ...

class AsyncIterator:
    def __init__(self, client, item_to_value, page_token=None, max_results=None): ...

Page Iteration

Protocol Buffer Utilities

Helper functions for working with Google's protocol buffer messages, including type conversion, field manipulation, and message introspection.

def from_any_pb(pb_type, any_pb): ...
def check_oneof(**kwargs): ...
def get_messages(module): ...
def get(msg_or_dict, key, default=None): ...
def set(msg_or_dict, key, value): ...
def field_mask(original, modified): ...

Protocol Buffer Helpers

Transport Helpers

Utilities for gRPC and REST transport protocols, including channel creation, method wrapping, and streaming support.

def create_channel(target, credentials=None, scopes=None, ssl_credentials=None, **kwargs): ...
def wrap_method(func, default_retry=None, default_timeout=None, client_info=None): ...

class GrpcStream:
    def __init__(self, wrapped): ...

Transport Helpers

Client Configuration

Configuration classes for customizing Google API client behavior, including authentication, endpoints, and transport options.

class ClientOptions:
    def __init__(self, api_endpoint=None, client_cert_source=None, client_encrypted_cert_source=None, **kwargs): ...

class ClientInfo:
    def __init__(self, client_library_name=None, client_library_version=None, **kwargs): ...

def from_dict(options): ...

Client Configuration

Date/Time Utilities

Utilities for handling datetime objects in Google APIs, including RFC3339 formatting, timezone handling, and high-precision timestamps.

class DatetimeWithNanoseconds(datetime.datetime):
    def __new__(cls, year, month, day, hour=0, minute=0, second=0, microsecond=0, nanosecond=0, **kwargs): ...

def utcnow(): ...
def from_rfc3339(value): ...
def to_rfc3339(value, ignore_zone=True): ...
def from_microseconds(value): ...
def to_microseconds(value): ...

Date/Time Utilities

Path Templates

URI template expansion and validation for Google API resource paths, supporting variable substitution and path parameter extraction.

def expand(tmpl, *args, **kwargs): ...
def validate(tmpl, path): ...
def get_field(request, field): ...
def delete_field(request, field): ...
def transcode(http_options, message=None, **request_kwargs): ...

Path Templates

Timeout Management

Timeout decorators and utilities for managing operation deadlines with various timeout strategies.

class TimeToDeadlineTimeout:
    def __init__(self, deadline): ...

class ConstantTimeout:
    def __init__(self, timeout): ...

class ExponentialTimeout:
    def __init__(self, initial_timeout, max_timeout, multiplier): ...

Timeout Management

Bidirectional Streaming

Support for bidirectional streaming gRPC operations with queue-based request generation, background consumers, and resumable streams.

class BidiRpc:
    def __init__(self, start_rpc, should_recover=None): ...

class ResumableBidiRpc(BidiRpc):
    def __init__(self, start_rpc, should_recover=None, should_terminate=None, throttle_reopen=None): ...

class BackgroundConsumer:
    def __init__(self, rpc, should_recover=None): ...

Bidirectional Streaming

IAM Policy Management

Non-API-specific IAM policy definitions for managing roles, permissions, and conditional bindings across Google Cloud resources.

class Policy:
    def __init__(self, bindings=None, etag=None, version=None): ...

class InvalidOperationException(ValueError): ...

# Policy binding format
def from_api_repr(resource): ...
def to_api_repr(policy): ...

IAM Policy Management

GAPIC Framework

Generated API client (GAPIC) infrastructure providing method wrapping, configuration parsing, and routing header generation for Google API clients.

def wrap_method(func, default_retry=None, default_timeout=None, client_info=None): ...
def wrap_async_method(func, default_retry=None, default_timeout=None, client_info=None): ...

def parse_method_configs(interface_config, client_config=None): ...
def to_routing_header(params): ...
def to_grpc_metadata(routing_header): ...

GAPIC Framework

Universe Domain Support

Universe domain configuration and validation for multi-environment Google Cloud deployments supporting custom endpoints and domain validation.

DEFAULT_UNIVERSE = "googleapis.com"

class EmptyUniverseError(ValueError): ...
class UniverseMismatchError(ValueError): ...

def determine_domain(client_universe_domain, universe_domain_env): ...
def get_universe_domain(client_universe_domain=None, universe_domain_env=None): ...

Universe Domain Support

Types

# Core type aliases
import typing
from typing import Any, Callable, Dict, List, Optional, Union

# Common type definitions used across modules
Predicate = Callable[[Exception], bool]
SleepGenerator = typing.Generator[float, None, None]