tessl/pypi-google-cloud-trace
Google Cloud Trace API client library for distributed tracing and performance analysis
—
Pending
v1-client.mddocs/
V1 Client Operations
Legacy trace operations for basic trace collection, retrieval, and management. The v1 API provides compatibility with older tracing implementations and supports fundamental trace operations with simple data structures.
Capabilities
Synchronous Client
The synchronous TraceServiceClient provides comprehensive v1 trace operations for managing traces and spans.
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
): ...
def list_traces(
self,
request: Optional[Union[ListTracesRequest, dict]] = None,
*,
project_id: Optional[str] = None,
view: Optional[ListTracesRequest.ViewType] = None,
page_size: Optional[int] = None,
start_time: Optional[Timestamp] = None,
end_time: Optional[Timestamp] = None,
filter: Optional[str] = None,
order_by: Optional[str] = None,
retry: OptionalRetry = None,
timeout: Union[float, object] = None,
metadata: Sequence[Tuple[str, str]] = ()
) -> ListTracesPager:
"""
Returns a list of traces that match the specified filter conditions.
Args:
request: The request object for listing traces
project_id: Required. The project ID where traces are stored
view: Type of data returned for traces (MINIMAL, ROOTSPAN, COMPLETE)
page_size: Maximum number of traces to return
start_time: Start of the time range for traces
end_time: End of the time range for traces
filter: Optional filter expression for traces
order_by: Field used to sort the returned traces
retry: Designation of what errors should be retried
timeout: The timeout for this request
metadata: Strings which should be sent along with the request as metadata
Returns:
Paginated list of traces matching the filter conditions
Raises:
google.api_core.exceptions.GoogleAPICallError: If the request failed
"""
def get_trace(
self,
request: Optional[Union[GetTraceRequest, dict]] = None,
*,
project_id: Optional[str] = None,
trace_id: Optional[str] = None,
retry: OptionalRetry = None,
timeout: Union[float, object] = None,
metadata: Sequence[Tuple[str, str]] = ()
) -> Trace:
"""
Gets a single trace by its ID.
Args:
request: The request object for getting a trace
project_id: Required. The project ID where the trace is stored
trace_id: Required. The ID of the trace to retrieve
retry: Designation of what errors should be retried
timeout: The timeout for this request
metadata: Strings which should be sent along with the request as metadata
Returns:
The requested trace with all its spans
Raises:
google.api_core.exceptions.GoogleAPICallError: If the request failed
google.api_core.exceptions.NotFound: If the trace doesn't exist
"""
def patch_traces(
self,
request: Optional[Union[PatchTracesRequest, dict]] = None,
*,
project_id: Optional[str] = None,
traces: Optional[Traces] = None,
retry: OptionalRetry = None,
timeout: Union[float, object] = None,
metadata: Sequence[Tuple[str, str]] = ()
) -> None:
"""
Sends new traces to Cloud Trace or updates existing traces.
Args:
request: The request object containing traces to patch
project_id: Required. The project ID where traces should be written
traces: Required. The body of the patch request containing traces
retry: Designation of what errors should be retried
timeout: The timeout for this request
metadata: Strings which should be sent along with the request as metadata
Raises:
google.api_core.exceptions.GoogleAPICallError: If the request failed
"""
@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
"""
@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
"""
# Alias for from_service_account_file
from_service_account_json = from_service_account_file
@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."""
def __enter__(self) -> "TraceServiceClient":
"""Enter the client context."""
def __exit__(self, type, value, traceback) -> None:
"""Exit the client context and clean up resources."""Asynchronous Client
The asynchronous TraceServiceAsyncClient provides the same functionality as the synchronous client but with async/await support.
class TraceServiceAsyncClient:
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
): ...
async def list_traces(
self,
request: Optional[Union[ListTracesRequest, dict]] = None,
*,
project_id: Optional[str] = None,
view: Optional[ListTracesRequest.ViewType] = None,
page_size: Optional[int] = None,
start_time: Optional[Timestamp] = None,
end_time: Optional[Timestamp] = None,
filter: Optional[str] = None,
order_by: Optional[str] = None,
retry: OptionalRetry = None,
timeout: Union[float, object] = None,
metadata: Sequence[Tuple[str, str]] = ()
) -> ListTracesAsyncPager:
"""
Returns a list of traces that match the specified filter conditions (async version).
Args:
request: The request object for listing traces
project_id: Required. The project ID where traces are stored
view: Type of data returned for traces (MINIMAL, ROOTSPAN, COMPLETE)
page_size: Maximum number of traces to return
start_time: Start of the time range for traces
end_time: End of the time range for traces
filter: Optional filter expression for traces
order_by: Field used to sort the returned traces
retry: Designation of what errors should be retried
timeout: The timeout for this request
metadata: Strings which should be sent along with the request as metadata
Returns:
Paginated async iterator of traces matching the filter conditions
Raises:
google.api_core.exceptions.GoogleAPICallError: If the request failed
"""
async def get_trace(
self,
request: Optional[Union[GetTraceRequest, dict]] = None,
*,
project_id: Optional[str] = None,
trace_id: Optional[str] = None,
retry: OptionalRetry = None,
timeout: Union[float, object] = None,
metadata: Sequence[Tuple[str, str]] = ()
) -> Trace:
"""
Gets a single trace by its ID (async version).
Args:
request: The request object for getting a trace
project_id: Required. The project ID where the trace is stored
trace_id: Required. The ID of the trace to retrieve
retry: Designation of what errors should be retried
timeout: The timeout for this request
metadata: Strings which should be sent along with the request as metadata
Returns:
The requested trace with all its spans
Raises:
google.api_core.exceptions.GoogleAPICallError: If the request failed
google.api_core.exceptions.NotFound: If the trace doesn't exist
"""
async def patch_traces(
self,
request: Optional[Union[PatchTracesRequest, dict]] = None,
*,
project_id: Optional[str] = None,
traces: Optional[Traces] = None,
retry: OptionalRetry = None,
timeout: Union[float, object] = None,
metadata: Sequence[Tuple[str, str]] = ()
) -> None:
"""
Sends new traces to Cloud Trace or updates existing traces (async version).
Args:
request: The request object containing traces to patch
project_id: Required. The project ID where traces should be written
traces: Required. The body of the patch request containing traces
retry: Designation of what errors should be retried
timeout: The timeout for this request
metadata: Strings which should be sent along with the request as metadata
Raises:
google.api_core.exceptions.GoogleAPICallError: If the request failed
"""
@classmethod
def from_service_account_file(
cls,
filename: str,
*args,
**kwargs
) -> "TraceServiceAsyncClient":
"""
Creates an async 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 TraceServiceAsyncClient instance
"""
@classmethod
def from_service_account_info(
cls,
info: Dict[str, Any],
*args,
**kwargs
) -> "TraceServiceAsyncClient":
"""
Creates an async 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 TraceServiceAsyncClient instance
"""
# Alias for from_service_account_file
from_service_account_json = from_service_account_file
@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."""
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."""Usage Examples
Listing Traces
from google.cloud import trace_v1
# Initialize client
client = trace_v1.TraceServiceClient()
# List traces for a project
request = trace_v1.ListTracesRequest(
project_id="my-project",
view=trace_v1.ListTracesRequest.ViewType.COMPLETE,
start_time={"seconds": 1609459200},
end_time={"seconds": 1609462800},
page_size=100
)
# Get traces
traces_pager = client.list_traces(request=request)
# Iterate through traces
for trace in traces_pager:
print(f"Trace ID: {trace.trace_id}")
print(f"Project: {trace.project_id}")
print(f"Number of spans: {len(trace.spans)}")Getting a Single Trace
from google.cloud import trace_v1
# Initialize client
client = trace_v1.TraceServiceClient()
# Get a specific trace
trace = client.get_trace(
project_id="my-project",
trace_id="abc123def456"
)
print(f"Trace ID: {trace.trace_id}")
for span in trace.spans:
print(f"Span: {span.name} ({span.span_id})")Creating/Updating Traces
from google.cloud import trace_v1
# Initialize client
client = trace_v1.TraceServiceClient()
# Create trace span
span = trace_v1.TraceSpan(
span_id=12345,
kind=trace_v1.TraceSpan.SpanKind.RPC_SERVER,
name="my-operation",
start_time={"seconds": 1609459200, "nanos": 500000000},
end_time={"seconds": 1609459201, "nanos": 750000000},
labels={
"component": "database",
"operation": "query"
}
)
# Create trace
trace = trace_v1.Trace(
project_id="my-project",
trace_id="abc123def456",
spans=[span]
)
# Send traces to Cloud Trace
traces = trace_v1.Traces(traces=[trace])
client.patch_traces(
project_id="my-project",
traces=traces
)Async Operations
import asyncio
from google.cloud import trace_v1
async def list_traces_async():
# Initialize async client
client = trace_v1.TraceServiceAsyncClient()
# List traces asynchronously
request = trace_v1.ListTracesRequest(
project_id="my-project",
view=trace_v1.ListTracesRequest.ViewType.MINIMAL,
page_size=50
)
traces_pager = await client.list_traces(request=request)
# Iterate through traces
async for trace in traces_pager:
print(f"Trace ID: {trace.trace_id}")
# Run async operation
asyncio.run(list_traces_async())Install with Tessl CLI
npx tessl i tessl/pypi-google-cloud-trace