grpc_utils

drunc.utils.grpc_utils

Classes

GrpcErrorDetails(code, message, details) dataclass

A structured representation of a gRPC error, including its status code, message, and any extracted rich error details. Used to extract and format detailed error information on the client side.

Attributes:

Name	Type	Description
code	str	The gRPC status code name (e.g., "NOT_FOUND")
message	str	The error message from the gRPC status
details	List[str]	A list of formatted error detail strings

Functions

__str__()

Return a human-readable string representation of the error.

Source code in drunc/utils/grpc_utils.py

def __str__(self):
    """
    Return a human-readable string representation of the error.
    """
    lines = [f"[{self.code}] {self.message}"]
    for detail in self.details:
        lines.append(f"{detail}")
    return "\n".join(lines)

RichErrorServerInterceptor

Bases: ServerInterceptor

A gRPC server interceptor that catches exceptions and converts them into rich error statuses with structured error details.

Functions

intercept_service(continuation, handler_call_details)

Intercept gRPC service calls to handle exceptions and convert them into rich error statuses.

Source code in drunc/utils/grpc_utils.py

def intercept_service(self, continuation, handler_call_details):
    """
    Intercept gRPC service calls to handle exceptions and convert them
    into rich error statuses.
    """
    handler = continuation(handler_call_details)

    def error_wrapper(request, context):
        try:
            return handler.unary_unary(request, context)

        except DruncSetupException as e:
            detail_obj = error_details_pb2.PreconditionFailure(
                violations=[
                    error_details_pb2.PreconditionFailure.Violation(
                        type="MISSING OR INVALID",
                        subject=str(e),
                        description=str(e.details),
                    )
                ]
            )
            abort_with_rich_error_status(
                context, e.grpc_error_code, str(e), detail_obj
            )

        except Exception as e:
            # Fallback
            detail = error_details_pb2.ErrorInfo(
                reason="Unexpected error",
                domain="server",
                metadata={"exception": str(type(e))},
            )
            return self._abort_with_detail(
                context, code_pb2.INTERNAL, str(e), detail
            )

    if handler.unary_unary:
        # only wrap unary-unary calls
        return grpc.unary_unary_rpc_method_handler(
            error_wrapper,
            request_deserializer=handler.request_deserializer,
            response_serializer=handler.response_serializer,
        )
    return handler

Functions

abort_with_rich_error_status(context, grpc_error_code, message, error_obj)

Aborts the current gRPC call with a rich error status containing structured error details.

Parameters:

Name	Type	Description	Default
context	ServicerContext	The gRPC context used to abort the RPC	required
grpc_error_code	Code	A gRPC status code from google.rpc.code_pb2 (e.g., code_pb2.INTERNAL, code_pb2.INVALID_ARGUMENT)	required
message	str	Quick description of the error	required
error_obj	Message	A protobuf message providing additional structured error details. It will be packed into a google.protobuf.Any	required

Raises: grpc.RpcError: Terminate the RPC with the constructed error status

Source code in drunc/utils/grpc_utils.py

def abort_with_rich_error_status(
    context: grpc.ServicerContext,
    grpc_error_code: code_pb2.Code,
    message: str,
    error_obj: Message,
) -> NoReturn:
    """
    Aborts the current gRPC call with a rich error status containing
    structured error details.

    Args:
        context (grpc.ServicerContext): The gRPC context used to abort the RPC
        grpc_error_code (code_pb2.Code): A gRPC status code from `google.rpc.code_pb2`
            (e.g., `code_pb2.INTERNAL`, `code_pb2.INVALID_ARGUMENT`)
        message (str): Quick description of the error
        error_obj (Message): A protobuf message providing additional structured
            error details. It will be packed into a google.protobuf.Any
    Raises:
        grpc.RpcError: Terminate the RPC with the constructed error status
    """

    detail_any = any_pb2.Any()
    detail_any.Pack(error_obj)

    rich_status = status_pb2.Status(
        code=grpc_error_code,
        message=message,
        details=[detail_any],
    )

    context.abort_with_status(rpc_status.to_status(rich_status))

copy_token(token)

Create a copy of the original token.

Parameters:

Name	Type	Description	Default
token	Token	The original token to copy.	required

Returns:

Type	Description
Token	A copy of the original token.

Source code in drunc/utils/grpc_utils.py

def copy_token(token: Token) -> Token:
    """Create a copy of the original token.

    Args:
        token: The original token to copy.

    Returns:
        A copy of the original token.
    """
    token_copy = Token()
    token_copy.CopyFrom(token)
    return token_copy

dict_to_grpc_proto(data, proto_class_instance)

Converts a Python dictionary into an instance of a gRPC Protobuf message. 'proto_class_instance' should be an empty instance, e.g., Token()

Source code in drunc/utils/grpc_utils.py

def dict_to_grpc_proto(data: dict, proto_class_instance: Message) -> Message:
    """
    Converts a Python dictionary into an instance of a gRPC Protobuf message.
    'proto_class_instance' should be an empty instance, e.g., Token()
    """
    return json_format.ParseDict(data, proto_class_instance, ignore_unknown_fields=True)

extract_grpc_rich_error(grpc_error)

Extract rich error details from a gRPC error using Google's error model.

Parameters:

Name	Type	Description	Default
grpc_error	RpcError	The gRPC error to parse	required

Returns:

Type	Description
GrpcErrorDetails	GrpcErrorDetails with structured error information

Source code in drunc/utils/grpc_utils.py

def extract_grpc_rich_error(grpc_error: grpc.RpcError) -> GrpcErrorDetails:
    """
    Extract rich error details from a gRPC error using Google's error model.

    Args:
        grpc_error: The gRPC error to parse

    Returns:
        GrpcErrorDetails with structured error information
    """
    code = grpc_error.code().name if grpc_error.code() else "UNKNOWN"
    try:
        status = rpc_status.from_call(grpc_error)
    except NotImplementedError:
        return GrpcErrorDetails(code=code, message="No message", details=[])

    # Fallback to simple error if no rich status
    if status is None:
        return GrpcErrorDetails(code=code, message="No message", details=[])

    # Extract all error details
    error_details = []
    for any_detail in status.details:
        detail_extracted = False
        for detail_type in _ERROR_DETAIL_TYPES:
            if any_detail.Is(detail_type.DESCRIPTOR):
                msg = detail_type()
                any_detail.Unpack(msg)
                error_details.extend(format_error_details(msg))
                detail_extracted = True
                break

        if not detail_extracted:
            error_details.append(f"Unknown detail type: {any_detail.type_url}")

    return GrpcErrorDetails(
        code=code, message=status.message or "No message", details=error_details
    )

format_error_details(detail)

Format protobuf message fields into human-readable strings.

Parameters:

Name	Type	Description	Default
detail	Message	A protobuf message representing a gRPC error detail	required

Returns:

Type	Description
list[str]	list[str]: A list of formatted strings describing the message's fields and values. Format: "field_name: value" for simple messages or "field_name: field1=value1, field2=value2" for nested messages

Source code in drunc/utils/grpc_utils.py

def format_error_details(detail: Message) -> list[str]:
    """
    Format protobuf message fields into human-readable strings.

    Args:
        detail (Message): A protobuf message representing a gRPC error detail

    Returns:
        list[str]: A list of formatted strings describing the message's fields and values.
                    Format: "field_name: value" for simple messages
                    or "field_name: field1=value1, field2=value2" for nested messages
    """

    results = []

    for field in detail.DESCRIPTOR.fields:
        value = getattr(detail, field.name)

        # Skip empty values
        if not value and value != 0 and value is not False:
            continue

        # Handle nested messages
        if field.type == FieldDescriptor.TYPE_MESSAGE:
            if field.is_repeated:
                # Handle repeated nested messages
                for item in value:
                    parts = _extract_message_parts(item)
                    if parts:
                        results.append(f"{field.name}: {', '.join(parts)}")
            else:
                # Handle single nested message
                parts = _extract_message_parts(value)
                if parts:
                    results.append(f"{field.name}: {', '.join(parts)}")
        else:
            # Handle simple fields
            results.append(f"{field.name}: {value}")

    return results if results else [str(detail)]

handle_grpc_error(error)

Handle gRPC errors by rethrowing them with appropriate context.

Parameters:

Name	Type	Description	Default
error	RpcError	The gRPC error to handle.	required

Raises: A custom exception if the error matches a known category, or the original gRPC error if no classification applies.

Source code in drunc/utils/grpc_utils.py

def handle_grpc_error(error: grpc.RpcError) -> NoReturn:
    """
    Handle gRPC errors by rethrowing them with appropriate context.

    Args:
        error: The gRPC error to handle.
    Raises:
        A custom exception if the error matches a known category, or the original
        gRPC error if no classification applies.
    """
    rethrow_if_unreachable_server(error)
    rethrow_if_timeout(error)
    raise error

interrupt_if_unreachable_server(grpc_error)

Interrupt if server is not reachable and return the error details.

Parameters:

Name	Type	Description	Default
grpc_error	RpcError	The gRPC error	required

Returns:

Type	Description
Optional[str]	str | None: The internal error details if the server is unreachable and details are available; otherwise, returns None.

Source code in drunc/utils/grpc_utils.py

def interrupt_if_unreachable_server(grpc_error: grpc.RpcError) -> Optional[str]:
    """
    Interrupt if server is not reachable and return the error details.

    Args:
        grpc_error (grpc.RpcError): The gRPC error

    Returns:
        str | None: The internal error details if the server is unreachable and details are available;
                    otherwise, returns None.
    """
    if not server_is_reachable(grpc_error):
        if hasattr(grpc_error, "_state"):
            return grpc_error._state.details
        elif hasattr(grpc_error, "_details"):
            return grpc_error._details

rethrow_if_timeout(grpc_error)

Raise a ServerTimeout if timeout.

Parameters:

Name	Type	Description	Default
grpc_error	RpcError	The gRPC error	required

Raises:

Type	Description
ServerTimeout	If the error code is DEADLINE_EXCEEDED

Source code in drunc/utils/grpc_utils.py

def rethrow_if_timeout(grpc_error: grpc.RpcError) -> NoReturn:
    """
    Raise a ServerTimeout if timeout.

    Args:
        grpc_error (grpc.RpcError): The gRPC error

    Raises:
        ServerTimeout: If the error code is DEADLINE_EXCEEDED
    """
    if hasattr(grpc_error, "_state"):
        if grpc_error._state.code == grpc.StatusCode.DEADLINE_EXCEEDED:
            raise ServerTimeout(grpc_error._state.details) from grpc_error

rethrow_if_unreachable_server(grpc_error)

Raise a ServerUnreachable exception if the gRPC error indicates the server is unreachable.

Parameters:

Name	Type	Description	Default
grpc_error	RpcError	The gRPC error	required

Raises:

Type	Description
ServerUnreachable	If the error indicates the server is unavailable

Source code in drunc/utils/grpc_utils.py

def rethrow_if_unreachable_server(grpc_error: grpc.RpcError) -> NoReturn:
    """
    Raise a ServerUnreachable exception if the gRPC error indicates the server is unreachable.

    Args:
        grpc_error (grpc.RpcError): The gRPC error

    Raises:
        ServerUnreachable: If the error indicates the server is unavailable
    """
    if not server_is_reachable(grpc_error):
        if hasattr(grpc_error, "_state"):
            raise ServerUnreachable(grpc_error._state.details) from grpc_error
        elif hasattr(grpc_error, "_details"):
            raise ServerUnreachable(grpc_error._details) from grpc_error

server_is_reachable(grpc_error)

Check if server is reachable.

Parameters:

Name	Type	Description	Default
grpc_error	RpcError	The gRPC error	required

Returns:

Name	Type	Description
bool	bool	True if the server is reachable, False if the error indicates it is unavailable

Source code in drunc/utils/grpc_utils.py

def server_is_reachable(grpc_error: grpc.RpcError) -> bool:
    """
    Check if server is reachable.

    Args:
        grpc_error (grpc.RpcError): The gRPC error

    Returns:
        bool: True if the server is reachable, False if the error indicates it is unavailable
    """
    if hasattr(grpc_error, "_state"):
        if grpc_error._state.code == grpc.StatusCode.UNAVAILABLE:
            return False

    elif hasattr(grpc_error, "_code"):
        if grpc_error._code == grpc.StatusCode.UNAVAILABLE:
            return False

    return True

unpack_error_response(name, text, token)

Create a response for unpacking errors.

Parameters:

Name	Type	Description	Default
name	str	The name of the command or service.	required
text	str	The error message to include in the response.	required
token	Token	The token associated with the request.	required

Returns:

Name	Type	Description
response	Response	the response object containing the error message.

Source code in drunc/utils/grpc_utils.py

def unpack_error_response(name: str, text: str, token: Token) -> Response:
    """Create a response for unpacking errors.

    Args:
        name: The name of the command or service.
        text: The error message to include in the response.
        token: The token associated with the request.

    Returns:
        response: the response object containing the error message.
    """
    return Response(
        name=name,
        token=token,
        data=pack_to_any(PlainText(text=text)),
        flag=ResponseFlag.NOT_EXECUTED_BAD_REQUEST_FORMAT,
        children=[],
    )