Google Cloud Trace API client library for distributed tracing and performance analysis
—
Client authentication, configuration options, and resource path management utilities for Google Cloud Trace. This covers credential handling, client configuration, and helper methods for constructing resource paths.
Both v1 and v2 clients support automatic credential discovery using Google Cloud's default credential chain.
class TraceServiceClient:
def __init__(
self,
*,
credentials: Optional[Credentials] = None,
transport: Optional[Union[str, TraceServiceTransport, Callable]] = None,
client_options: Optional[Union[ClientOptions, dict]] = None,
client_info: gapic_v1.client_info.ClientInfo = None
): ...Create clients using service account credentials from file or dictionary.
class TraceServiceClient:
@classmethod
def from_service_account_file(
cls,
filename: str,
*args,
**kwargs
) -> TraceServiceClient:
"""
Creates a client instance from a service account JSON file.
Args:
filename: Path to the service account JSON file
*args: Additional arguments to pass to the client constructor
**kwargs: Additional keyword arguments to pass to the client constructor
Returns:
Constructed TraceServiceClient instance
Raises:
google.auth.exceptions.DefaultCredentialsError: If the file is not found or invalid
"""
@classmethod
def from_service_account_info(
cls,
info: Dict[str, Any],
*args,
**kwargs
) -> TraceServiceClient:
"""
Creates a client instance from service account information.
Args:
info: Service account information as a dictionary
*args: Additional arguments to pass to the client constructor
**kwargs: Additional keyword arguments to pass to the client constructor
Returns:
Constructed TraceServiceClient instance
Raises:
google.auth.exceptions.DefaultCredentialsError: If the info is invalid
"""
# Alias for from_service_account_file
from_service_account_json = from_service_account_fileUtility methods for constructing properly formatted resource names.
class TraceServiceClient:
@staticmethod
def common_project_path(project: str) -> str:
"""
Returns a fully-qualified project string.
Args:
project: Project ID
Returns:
Formatted project resource name: "projects/{project}"
"""
@staticmethod
def parse_common_project_path(path: str) -> Dict[str, str]:
"""
Parses a project path into its component segments.
Args:
path: Project resource path
Returns:
Dictionary with parsed path components
"""
@staticmethod
def common_location_path(project: str, location: str) -> str:
"""
Returns a fully-qualified location string.
Args:
project: Project ID
location: Location ID
Returns:
Formatted location resource name: "projects/{project}/locations/{location}"
"""
@staticmethod
def parse_common_location_path(path: str) -> Dict[str, str]:
"""
Parses a location path into its component segments.
Args:
path: Location resource path
Returns:
Dictionary with keys: project, location
"""
@staticmethod
def common_billing_account_path(billing_account: str) -> str:
"""
Returns a fully-qualified billing account string.
Args:
billing_account: Billing account ID
Returns:
Formatted billing account resource name: "billingAccounts/{billing_account}"
"""
@staticmethod
def parse_common_billing_account_path(path: str) -> Dict[str, str]:
"""
Parses a billing account path into its component segments.
Args:
path: Billing account resource path
Returns:
Dictionary with keys: billing_account
"""
@staticmethod
def common_folder_path(folder: str) -> str:
"""
Returns a fully-qualified folder string.
Args:
folder: Folder ID
Returns:
Formatted folder resource name: "folders/{folder}"
"""
@staticmethod
def parse_common_folder_path(path: str) -> Dict[str, str]:
"""
Parses a folder path into its component segments.
Args:
path: Folder resource path
Returns:
Dictionary with keys: folder
"""
@staticmethod
def common_organization_path(organization: str) -> str:
"""
Returns a fully-qualified organization string.
Args:
organization: Organization ID
Returns:
Formatted organization resource name: "organizations/{organization}"
"""
@staticmethod
def parse_common_organization_path(path: str) -> Dict[str, str]:
"""
Parses an organization path into its component segments.
Args:
path: Organization resource path
Returns:
Dictionary with keys: organization
"""Resource path helpers specific to the v2 API for span management.
class TraceServiceClient:
@staticmethod
def span_path(project: str, trace: str, span: str) -> str:
"""
Returns a fully-qualified span string.
Args:
project: Project ID
trace: Trace ID
span: Span ID
Returns:
Formatted span resource name: "projects/{project}/traces/{trace}/spans/{span}"
"""
@staticmethod
def parse_span_path(path: str) -> Dict[str, str]:
"""
Parses a span path into its component segments.
Args:
path: Span resource path
Returns:
Dictionary with keys: project, trace, span
"""Configure transport mechanisms for client communication.
# Transport types
TransportType = Union[str, TraceServiceTransport, Callable]
# Available transport strings:
# - "grpc" (default): gRPC transport
# - "grpc_asyncio": Async gRPC transport
# - "rest": REST transportConfigure client behavior through ClientOptions.
class ClientOptions:
api_endpoint: Optional[str] # Override default API endpoint
client_cert_source: Optional[Callable] # Client certificate source
quota_project_id: Optional[str] # Project for quota and billing
credentials_file: Optional[str] # Path to credentials file
scopes: Optional[Sequence[str]] # OAuth 2.0 scopes
default_scopes: Optional[Sequence[str]] # Default OAuth 2.0 scopesAccess client configuration and transport information.
class TraceServiceClient:
@property
def transport(self) -> TraceServiceTransport:
"""Returns the transport used by the client instance."""
@property
def api_endpoint(self) -> str:
"""Returns the API endpoint used by the client instance."""
@property
def universe_domain(self) -> str:
"""Returns the universe domain used by the client instance."""Both synchronous and asynchronous clients support context management for automatic resource cleanup.
class TraceServiceClient:
def __enter__(self) -> "TraceServiceClient":
"""Enter the client context."""
def __exit__(self, type, value, traceback) -> None:
"""Exit the client context and clean up resources."""
class TraceServiceAsyncClient:
async def __aenter__(self) -> "TraceServiceAsyncClient":
"""Enter the async client context."""
async def __aexit__(self, type, value, traceback) -> None:
"""Exit the async client context and clean up resources."""from google.cloud import trace_v2
# Using default credentials (recommended)
client = trace_v2.TraceServiceClient()
# Using specific credentials
from google.oauth2 import service_account
credentials = service_account.Credentials.from_service_account_file(
"path/to/service-account.json"
)
client = trace_v2.TraceServiceClient(credentials=credentials)from google.cloud import trace_v2
# From service account file
client = trace_v2.TraceServiceClient.from_service_account_file(
"path/to/credentials.json"
)
# From service account info dictionary
service_account_info = {
"type": "service_account",
"project_id": "my-project",
"private_key_id": "key-id",
"private_key": "-----BEGIN PRIVATE KEY-----\n...",
"client_email": "service@my-project.iam.gserviceaccount.com",
"client_id": "123456789",
"auth_uri": "https://accounts.google.com/o/oauth2/auth",
"token_uri": "https://oauth2.googleapis.com/token"
}
client = trace_v2.TraceServiceClient.from_service_account_info(
service_account_info
)from google.cloud import trace_v2
from google.api_core import client_options
# Configure client options
options = client_options.ClientOptions(
api_endpoint="https://custom-trace-endpoint.com",
quota_project_id="billing-project"
)
client = trace_v2.TraceServiceClient(client_options=options)
# Configure transport
client = trace_v2.TraceServiceClient(transport="rest")
# Custom gRPC configuration
import grpc
from google.cloud.trace_v2.services.trace_service import transports
channel = grpc.secure_channel(
"cloudtrace.googleapis.com:443",
credentials=grpc.ssl_channel_credentials()
)
transport = transports.TraceServiceGrpcTransport(channel=channel)
client = trace_v2.TraceServiceClient(transport=transport)from google.cloud import trace_v2
# Create client
client = trace_v2.TraceServiceClient()
# Construct resource paths
project_path = client.common_project_path("my-project")
print(project_path) # "projects/my-project"
span_path = client.span_path("my-project", "trace-123", "span-456")
print(span_path) # "projects/my-project/traces/trace-123/spans/span-456"
# Parse resource paths
parsed = client.parse_span_path(span_path)
print(parsed) # {"project": "my-project", "trace": "trace-123", "span": "span-456"}
# Use in API calls
span = trace_v2.Span(
name=span_path,
span_id="span-456",
display_name=trace_v2.TruncatableString(value="my-operation"),
start_time={"seconds": 1609459200},
end_time={"seconds": 1609459201}
)
client.create_span(request=span)import os
from google.cloud import trace_v2
# Set credentials via environment variable
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "/path/to/credentials.json"
# Set project ID via environment variable
os.environ["GOOGLE_CLOUD_PROJECT"] = "my-project"
# Client will automatically use these credentials
client = trace_v2.TraceServiceClient()
# Use project from environment
project_name = client.common_project_path(os.environ["GOOGLE_CLOUD_PROJECT"])from google.cloud import trace_v2
from google.api_core import exceptions
try:
client = trace_v2.TraceServiceClient()
span = trace_v2.Span(
name="projects/my-project/traces/trace-123/spans/span-456",
span_id="span-456",
display_name=trace_v2.TruncatableString(value="test-span"),
start_time={"seconds": 1609459200},
end_time={"seconds": 1609459201}
)
client.create_span(request=span)
except exceptions.Unauthenticated:
print("Authentication failed - check credentials")
except exceptions.PermissionDenied:
print("Insufficient permissions for the operation")
except exceptions.InvalidArgument as e:
print(f"Invalid request parameters: {e}")
except exceptions.GoogleAPICallError as e:
print(f"API call failed: {e}")Install with Tessl CLI
npx tessl i tessl/pypi-google-cloud-trace