RPC Status and Error Handling

from google.rpc.status_pb2 import Status

class Status(message.Message):
    """RPC status with error details."""
    code: int  # Status code (typically from Code enum)
    message: str  # Human-readable error message
    details: list[Any]  # Additional error details

from google.rpc.code_pb2 import Code

class Code(enum.Enum):
    """Standard gRPC status codes."""
    OK = 0  # Success
    CANCELLED = 1  # Operation cancelled
    UNKNOWN = 2  # Unknown error
    INVALID_ARGUMENT = 3  # Invalid argument
    DEADLINE_EXCEEDED = 4  # Deadline exceeded
    NOT_FOUND = 5  # Resource not found
    ALREADY_EXISTS = 6  # Resource already exists
    PERMISSION_DENIED = 7  # Permission denied
    RESOURCE_EXHAUSTED = 8  # Resource exhausted
    FAILED_PRECONDITION = 9  # Failed precondition
    ABORTED = 10  # Operation aborted
    OUT_OF_RANGE = 11  # Out of range
    UNIMPLEMENTED = 12  # Unimplemented
    INTERNAL = 13  # Internal error
    UNAVAILABLE = 14  # Service unavailable
    DATA_LOSS = 15  # Data loss
    UNAUTHENTICATED = 16  # Unauthenticated

from google.rpc.error_details_pb2 import (
    ErrorInfo, RetryInfo, DebugInfo, QuotaFailure,
    PreconditionFailure, BadRequest, RequestInfo,
    ResourceInfo, Help, LocalizedMessage
)

class ErrorInfo(message.Message):
    """Additional error information."""
    reason: str  # Error reason identifier
    domain: str  # Error domain
    metadata: dict[str, str]  # Additional metadata

class RetryInfo(message.Message):
    """Retry timing information."""
    retry_delay: Duration  # Recommended retry delay

class DebugInfo(message.Message):
    """Debug information for development."""
    stack_entries: list[str]  # Stack trace entries
    detail: str  # Additional debug details

class QuotaFailure(message.Message):
    """Quota failure details."""
    violations: list[QuotaFailure.Violation]  # Quota violations

    class Violation(message.Message):
        """Individual quota violation."""
        subject: str  # Subject that violated quota
        description: str  # Violation description

class PreconditionFailure(message.Message):
    """Precondition failure details."""
    violations: list[PreconditionFailure.Violation]  # Precondition violations

    class Violation(message.Message):
        """Individual precondition violation."""
        type: str  # Violation type
        subject: str  # Subject of violation
        description: str  # Violation description

class BadRequest(message.Message):
    """Bad request details."""
    field_violations: list[BadRequest.FieldViolation]  # Field violations

    class FieldViolation(message.Message):
        """Individual field violation."""
        field: str  # Field path
        description: str  # Violation description

class RequestInfo(message.Message):
    """Request information for debugging."""
    request_id: str  # Unique request identifier
    serving_data: str  # Server-side data

class ResourceInfo(message.Message):
    """Resource information for errors."""
    resource_type: str  # Resource type identifier
    resource_name: str  # Resource name
    owner: str  # Resource owner
    description: str  # Resource description

class Help(message.Message):
    """Help information for error resolution."""
    links: list[Help.Link]  # Help links

    class Link(message.Message):
        """Individual help link."""
        description: str  # Link description
        url: str  # Help URL

class LocalizedMessage(message.Message):
    """Localized error message."""
    locale: str  # Message locale
    message: str  # Localized message text

from google.rpc.context.attribute_context_pb2 import AttributeContext
from google.rpc.context.audit_context_pb2 import AuditContext

class AttributeContext(message.Message):
    """Request attribute context."""
    source: AttributeContext.Peer  # Request source
    destination: AttributeContext.Peer  # Request destination
    request: AttributeContext.Request  # Request details
    response: AttributeContext.Response  # Response details
    resource: AttributeContext.Resource  # Target resource
    api: AttributeContext.Api  # API information

    class Peer(message.Message):
        """Network peer information."""
        ip: str  # IP address
        port: int  # Port number
        labels: dict[str, str]  # Peer labels
        principal: str  # Authenticated principal
        region_code: str  # Geographic region

    class Api(message.Message):
        """API information."""
        service: str  # Service name
        operation: str  # Operation name
        protocol: str  # Protocol used
        version: str  # API version

    class Auth(message.Message):
        """Authentication information."""
        principal: str  # Authenticated principal
        audiences: list[str]  # Token audiences
        presenter: str  # Token presenter
        claims: Struct  # Additional claims
        access_levels: list[str]  # Access levels

    class Request(message.Message):
        """Request information."""
        id: str  # Request ID
        method: str  # HTTP method
        headers: dict[str, str]  # Request headers
        path: str  # Request path
        host: str  # Request host
        scheme: str  # Request scheme
        query: str  # Query string
        time: Timestamp  # Request time
        size: int  # Request size
        protocol: str  # Protocol version
        reason: str  # Request reason
        auth: AttributeContext.Auth  # Authentication info

class AuditContext(message.Message):
    """Audit context information."""
    audit_log_name: str  # Audit log name
    request_metadata: RequestMetadata  # Request metadata
    service_data: Any  # Service-specific data

from google.rpc.http_pb2 import HttpRequest, HttpResponse

class HttpRequest(message.Message):
    """HTTP request representation."""
    method: str  # HTTP method
    uri: str  # Request URI  
    headers: list[HttpHeader]  # HTTP headers
    body: bytes  # Request body
    
class HttpResponse(message.Message):
    """HTTP response representation."""
    status: int  # HTTP status code
    headers: list[HttpHeader]  # Response headers
    body: bytes  # Response body

class HttpHeader(message.Message):
    """HTTP header representation."""
    name: str  # Header name
    value: str  # Header value

from google.rpc.status_pb2 import Status
from google.rpc.code_pb2 import Code

# Create a status for invalid argument
status = Status()
status.code = Code.INVALID_ARGUMENT
status.message = "The 'name' field is required but was not provided"

from google.rpc.error_details_pb2 import ErrorInfo, BadRequest
from google.protobuf.any_pb2 import Any

# Create detailed error information
error_info = ErrorInfo()
error_info.reason = "REQUIRED_FIELD_MISSING"
error_info.domain = "googleapis.com"
error_info.metadata["field"] = "name"

# Create field violation details
bad_request = BadRequest()
violation = bad_request.field_violations.add()
violation.field = "name"
violation.description = "This field is required"

# Pack into status details
status = Status()
status.code = Code.INVALID_ARGUMENT
status.message = "Request contains invalid arguments"

# Add error details
detail1 = Any()
detail1.Pack(error_info)
status.details.append(detail1)

detail2 = Any()
detail2.Pack(bad_request)
status.details.append(detail2)

from google.rpc.error_details_pb2 import RetryInfo
from google.protobuf.duration_pb2 import Duration

# Create retry information
retry_info = RetryInfo()
retry_info.retry_delay.seconds = 30  # Retry after 30 seconds

# Pack into status
status = Status()
status.code = Code.RESOURCE_EXHAUSTED
status.message = "Quota exceeded, retry after delay"

detail = Any()
detail.Pack(retry_info)
status.details.append(detail)

from google.rpc.status_pb2 import Status
from google.rpc.code_pb2 import Code
from google.rpc.error_details_pb2 import ErrorInfo, RetryInfo

def handle_rpc_error(status: Status):
    """Handle RPC error with detailed information."""
    print(f"RPC failed with code {status.code}: {status.message}")
    
    # Process error details
    for detail in status.details:
        if detail.Is(ErrorInfo.DESCRIPTOR):
            error_info = ErrorInfo()
            detail.Unpack(error_info)
            print(f"Error reason: {error_info.reason}")
            print(f"Error domain: {error_info.domain}")
            
        elif detail.Is(RetryInfo.DESCRIPTOR):
            retry_info = RetryInfo()
            detail.Unpack(retry_info)
            print(f"Retry after: {retry_info.retry_delay.seconds} seconds")

from google.rpc.error_details_pb2 import QuotaFailure

def handle_quota_failure(status: Status):
    """Handle quota failure with violation details."""
    for detail in status.details:
        if detail.Is(QuotaFailure.DESCRIPTOR):
            quota_failure = QuotaFailure()
            detail.Unpack(quota_failure)
            
            for violation in quota_failure.violations:
                print(f"Quota violation: {violation.subject}")
                print(f"Description: {violation.description}")