Google Cloud Redis API client library for managing fully managed Redis instances on Google Cloud
—
Core client functionality for authentication, configuration, and connection management to the Google Cloud Redis API.
The CloudRedisClient provides synchronous access to the Cloud Redis API with automatic authentication and connection management.
class CloudRedisClient:
def __init__(
self,
*,
credentials: Optional[ga_credentials.Credentials] = None,
transport: Optional[Union[str, CloudRedisTransport, Callable]] = None,
client_options: Optional[Union[ClientOptions, dict]] = None,
client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
) -> None:
"""
Initialize the Cloud Redis client.
Args:
credentials: The authorization credentials to attach to requests
transport: The transport to use for communication ('grpc', 'rest', or transport instance)
client_options: Client configuration options (endpoint, scopes, etc.)
client_info: Client information for user agent strings
Raises:
google.auth.exceptions.MutualTLSChannelError: If mutual TLS transport creation fails
"""The CloudRedisAsyncClient provides asynchronous access with the same interface as the synchronous client.
class CloudRedisAsyncClient:
def __init__(
self,
*,
credentials: Optional[ga_credentials.Credentials] = None,
transport: Optional[Union[str, CloudRedisTransport, Callable]] = None,
client_options: Optional[Union[ClientOptions, dict]] = None,
client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
) -> None:
"""
Initialize the async Cloud Redis client.
Args:
credentials: The authorization credentials to attach to requests
transport: The transport to use ('grpc_asyncio', 'rest', or transport instance)
client_options: Client configuration options
client_info: Client information for user agent strings
"""Create clients from service account credentials stored in files or dictionaries.
@classmethod
def from_service_account_info(
cls,
info: dict,
**kwargs
) -> CloudRedisClient:
"""
Create a client from service account info dictionary.
Args:
info: Service account info in Google format
**kwargs: Additional arguments for client initialization
Returns:
CloudRedisClient: The constructed client
"""
@classmethod
def from_service_account_file(
cls,
filename: str,
**kwargs
) -> CloudRedisClient:
"""
Create a client from a service account json file.
Args:
filename: Path to the service account json file
**kwargs: Additional arguments for client initialization
Returns:
CloudRedisClient: The constructed client
"""Access and configure the underlying transport layer for advanced use cases.
@classmethod
def get_transport_class(
cls,
label: Optional[str] = None,
) -> Type[CloudRedisTransport]:
"""
Return an appropriate transport class.
Args:
label: The name of the desired transport ('grpc', 'rest')
Returns:
Type[CloudRedisTransport]: The transport class
"""
@property
def transport(self) -> CloudRedisTransport:
"""
Access the transport used by the client.
Returns:
CloudRedisTransport: The transport instance
"""
@property
def api_endpoint(self) -> str:
"""
The API endpoint used by the client.
Returns:
str: The API endpoint URL
"""
@property
def universe_domain(self) -> str:
"""
The universe domain used by the client.
Returns:
str: The universe domain
"""Utility methods for constructing and parsing Google Cloud resource paths.
@staticmethod
def instance_path(project: str, location: str, instance: str) -> str:
"""
Return a fully-qualified instance string.
Args:
project: Project ID
location: Location/region
instance: Instance ID
Returns:
str: The instance resource path
"""
@staticmethod
def parse_instance_path(path: str) -> Dict[str, str]:
"""
Parse an instance path into its component segments.
Args:
path: Instance resource path
Returns:
Dict[str, str]: Dictionary with 'project', 'location', 'instance' keys
"""
@staticmethod
def common_billing_account_path(billing_account: str) -> str:
"""Return a billing account string."""
@staticmethod
def common_folder_path(folder: str) -> str:
"""Return a folder string."""
@staticmethod
def common_organization_path(organization: str) -> str:
"""Return an organization string."""
@staticmethod
def common_project_path(project: str) -> str:
"""Return a project string."""
@staticmethod
def common_location_path(project: str, location: str) -> str:
"""Return a location string."""Both client classes support context manager protocol for automatic cleanup.
def __enter__(self) -> CloudRedisClient:
"""Enter context manager."""
def __exit__(self, type, value, traceback) -> None:
"""Exit context manager and cleanup resources."""from google.cloud.redis import CloudRedisClient
# Use default credentials (ADC)
client = CloudRedisClient()
# Use service account file
client = CloudRedisClient.from_service_account_file(
"path/to/service-account.json"
)
# Use service account info
service_account_info = {
"type": "service_account",
"project_id": "your-project",
# ... other service account fields
}
client = CloudRedisClient.from_service_account_info(service_account_info)from google.cloud.redis import CloudRedisClient
from google.cloud.redis_v1.services.cloud_redis.transports import CloudRedisRestTransport
# Use REST transport instead of gRPC
transport = CloudRedisRestTransport()
client = CloudRedisClient(transport=transport)
# Configure transport with custom options
from google.api_core import client_options
options = client_options.ClientOptions(
api_endpoint="https://redis.googleapis.com",
scopes=["https://www.googleapis.com/auth/cloud-platform"]
)
client = CloudRedisClient(client_options=options)import asyncio
from google.cloud.redis import CloudRedisAsyncClient
async def main():
async with CloudRedisAsyncClient() as client:
# Use async client methods
parent = "projects/my-project/locations/us-central1"
instances = await client.list_instances(parent=parent)
async for instance in instances:
print(f"Instance: {instance.name}")
asyncio.run(main())from google.cloud.redis import CloudRedisClient
# Build resource paths
instance_path = CloudRedisClient.instance_path(
project="my-project",
location="us-central1",
instance="my-redis-instance"
)
print(instance_path) # "projects/my-project/locations/us-central1/instances/my-redis-instance"
# Parse resource paths
path_components = CloudRedisClient.parse_instance_path(instance_path)
print(path_components) # {'project': 'my-project', 'location': 'us-central1', 'instance': 'my-redis-instance'}Install with Tessl CLI
npx tessl i tessl/pypi-google-cloud-redis