tessl/pypi-google-api-core

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

—

Pending

Overview

Eval results

Files

GAPIC Framework

Name: tessl/pypi-google-api-core
Author: tessl

Generated API Client (GAPIC) infrastructure providing method wrapping, configuration parsing, and enhanced client functionality for Google API clients. This framework transforms low-level gRPC methods into high-level client library methods with automatic retry, timeout, compression, and error handling capabilities.

Capabilities

Method Wrapping

Core functionality for wrapping RPC methods with common behaviors including retry logic, timeout handling, compression, and error mapping.

def wrap_method(
    func,
    default_retry=None,
    default_timeout=None, 
    default_compression=None,
    client_info=None,
    *,
    with_call=False
): ...

Async Method Wrapping

Asynchronous version of method wrapping for async/await based client libraries.

def wrap_async_method(
    func,
    default_retry=None,
    default_timeout=None,
    default_compression=None,
    client_info=None,
    *,
    with_call=False
): ...

Configuration Parsing

Parse and process GAPIC configuration data for retry policies, timeout settings, and error handling rules.

def parse_method_configs(interface_config, client_config=None, calling_form=None): ...

def _retry_from_retry_config(retry_params, retry_codes, retry_impl=None): ...

def _timeout_from_retry_config(retry_params, timeout_impl=None): ...

Routing Headers

Generate routing headers for Google infrastructure to determine request routing, especially for regional services.

def to_routing_header(params, qualified_enums=True): ...

def to_grpc_metadata(routing_header): ...

ROUTING_METADATA_KEY = "x-goog-request-params"

Enhanced Client Information

Extended client information for API analytics including library versions, runtime information, and user agent details.

class ClientInfo:
    def __init__(
        self,
        python_version=None,
        grpc_version=None, 
        api_core_version=None,
        gapic_version=None,
        client_library_version=None,
        user_agent=None,
        rest_version=None,
        protobuf_runtime_version=None
    ): ...
    
    def to_user_agent(self): ...
    def to_grpc_metadata(self): ...

METRICS_METADATA_KEY = "x-goog-api-client"
DEFAULT_CLIENT_INFO = ClientInfo()

Usage Examples

Basic Method Wrapping

from google.api_core.gapic_v1 import method
from google.api_core import retry
from google.api_core import timeout
from grpc import Compression

# Original RPC method
def get_topic(name, timeout=None):
    request = publisher_v2.GetTopicRequest(name=name)
    return publisher_stub.GetTopic(request, timeout=timeout)

# Configure defaults
default_retry = retry.Retry(deadline=60)
default_timeout = timeout.Timeout(deadline=60) 
default_compression = Compression.NoCompression

# Wrap the method
wrapped_get_topic = method.wrap_method(
    get_topic,
    default_retry=default_retry,
    default_timeout=default_timeout,
    default_compression=default_compression
)

# Use wrapped method with defaults
response = wrapped_get_topic(name="projects/my-project/topics/my-topic")

# Override retry behavior
custom_retry = retry.Retry(
    predicate=retry.if_exception_type(exceptions.InternalServerError)
)
response = wrapped_get_topic(
    name="projects/my-project/topics/my-topic",
    retry=custom_retry
)

# Disable retry
response = wrapped_get_topic(
    name="projects/my-project/topics/my-topic", 
    retry=None
)

Configuration Parsing

from google.api_core.gapic_v1 import config

# Parse method configurations from GAPIC config
interface_config = {
    "retry_codes": {
        "idempotent": ["DEADLINE_EXCEEDED", "UNAVAILABLE"],
        "non_idempotent": []
    },
    "retry_params": {
        "default": {
            "initial_retry_delay_millis": 100,
            "retry_delay_multiplier": 1.3,
            "max_retry_delay_millis": 60000,
            "initial_rpc_timeout_millis": 20000,
            "rpc_timeout_multiplier": 1.0,
            "max_rpc_timeout_millis": 20000,
            "total_timeout_millis": 600000
        }
    },
    "methods": {
        "GetTopic": {
            "timeout_millis": 60000,
            "retry_codes_name": "idempotent",
            "retry_params_name": "default"
        }
    }
}

# Parse configurations
method_configs = config.parse_method_configs(interface_config)
get_topic_config = method_configs["GetTopic"]

# Access parsed retry and timeout objects
retry_policy = get_topic_config.retry
timeout_policy = get_topic_config.timeout

Routing Headers

from google.api_core.gapic_v1 import routing_header

# Create routing parameters
params = {
    "project": "my-project",
    "location": "us-central1",
    "instance": "my-instance"
}

# Generate routing header string
header_value = routing_header.to_routing_header(params)
# Returns: "project=my-project&location=us-central1&instance=my-instance"

# Convert to gRPC metadata
metadata = routing_header.to_grpc_metadata(header_value)
# Returns: [("x-goog-request-params", "project=my-project&location=us-central1&instance=my-instance")]

Client Information

from google.api_core.gapic_v1 import client_info

# Create client info with version details
info = client_info.ClientInfo(
    gapic_version="1.2.0",
    client_library_version="2.1.0",
    user_agent="my-application/1.0.0"
)

# Generate user agent string
user_agent = info.to_user_agent()
# Returns: "my-application/1.0.0 gapic/1.2.0 gax/2.1.0 gl-python/3.9.6"

# Convert to gRPC metadata
metadata = info.to_grpc_metadata()
# Returns: [("x-goog-api-client", "my-application/1.0.0 gapic/1.2.0 gax/2.1.0 gl-python/3.9.6")]

Method Configuration Defaults

from google.api_core.gapic_v1.method import DEFAULT

# Use sentinel value for default behavior
wrapped_method = method.wrap_method(
    original_method,
    default_retry=retry.Retry(),
    default_timeout=timeout.Timeout(60)
)

# Call with explicit defaults
response = wrapped_method(
    request,
    retry=DEFAULT,  # Use the default retry policy
    timeout=DEFAULT,  # Use the default timeout
    compression=DEFAULT  # Use default compression
)

Framework Architecture

Method Decoration Pipeline

The GAPIC framework applies decorators in a specific order to transform raw RPC methods:

Error Wrapping: Maps gRPC status codes to Google API exceptions
Retry Logic: Applies configurable retry policies with exponential backoff
Timeout Handling: Enforces method-level and call-level timeouts
Compression: Applies gRPC compression settings
Metadata Injection: Adds client info and routing headers
Instrumentation: Provides call tracing and metrics collection

Configuration Structure

GAPIC configurations define behavior policies for each RPC method:

{
    "retry_codes": {
        "policy_name": ["STATUS_CODE1", "STATUS_CODE2"]
    },
    "retry_params": {
        "policy_name": {
            "initial_retry_delay_millis": 100,
            "retry_delay_multiplier": 1.3,
            "max_retry_delay_millis": 60000,
            "total_timeout_millis": 600000
        }
    },
    "methods": {
        "MethodName": {
            "retry_codes_name": "policy_name",
            "retry_params_name": "policy_name"
        }
    }
}

Async Support

Full support for asyncio-based clients with async method wrapping:

Async Retry: Non-blocking retry logic using asyncio.sleep()
Async Timeout: Timeout handling compatible with asyncio event loops
Context Management: Proper resource cleanup in async contexts
Exception Propagation: Maintains exception semantics across async boundaries

Import Patterns

# Method wrapping
from google.api_core.gapic_v1 import method
from google.api_core.gapic_v1 import method_async

# Configuration parsing
from google.api_core.gapic_v1 import config

# Routing headers
from google.api_core.gapic_v1 import routing_header

# Client information
from google.api_core.gapic_v1 import client_info

# Access all modules
from google.api_core import gapic_v1

Types

from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union
import enum
import grpc

# Method wrapping types
WrappedMethod = Callable[..., Any]
GrpcMethod = Callable[..., Any]
Decorator = Callable[[Callable], Callable]

# Configuration types
RetryConfig = Dict[str, Union[str, int, float]]
MethodConfig = Dict[str, Union[str, RetryConfig]]
InterfaceConfig = Dict[str, Union[Dict, List]]

# Routing types
RoutingParams = Dict[str, Union[str, bytes, enum.Enum]]
GrpcMetadata = List[Tuple[str, str]]

# Default sentinel
class _MethodDefault(enum.Enum):
    _DEFAULT_VALUE = object()

DEFAULT = _MethodDefault._DEFAULT_VALUE

Constants and Metadata Keys

# Routing header metadata key
ROUTING_METADATA_KEY = "x-goog-request-params"

# Client metrics metadata key  
METRICS_METADATA_KEY = "x-goog-api-client"

# Cache size for routing parameter computation
ROUTING_PARAM_CACHE_SIZE = 32

# Milliseconds per second conversion
_MILLIS_PER_SECOND = 1000.0

# Sentinel for default metadata behavior
USE_DEFAULT_METADATA = object()

The GAPIC framework is the foundation for all generated Google API client libraries, providing consistent behavior, configuration management, and enhanced functionality across hundreds of Google Cloud and API services.

Install with Tessl CLI