Skip to content

Errors

The helpers/errors subpackage turns exceptions into transport responses. capture, validation and mapping are transport-agnostic; fastapi and grpc need the matching extra.

Capture

Capture exceptions: log them locally and record them on the active OpenTelemetry span.

archipy.helpers.errors.capture.logger module-attribute 

logger = logging.getLogger(
    "archipy.helpers.utils.error_utils"
)

archipy.helpers.errors.capture.capture_exception 

capture_exception(exception: BaseException) -> None

Captures an exception and records it on the current OpenTelemetry span.

Always logs locally. When OTel is enabled and a recording span is active, records the exception event and sets span status via OtelUtils.status_for_exception.

Parameters:

Name Type Description Default
exception BaseException

The exception to capture and report.

 required
Source code in archipy/helpers/errors/capture.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def capture_exception(exception: BaseException) -> None:
    """Captures an exception and records it on the current OpenTelemetry span.

    Always logs locally. When OTel is enabled and a recording span is active,
    records the exception event and sets span status via ``OtelUtils.status_for_exception``.

    Args:
        exception (BaseException): The exception to capture and report.
    """
    logger.error(
        "An exception occurred: %s: %s",
        type(exception).__name__,
        exception,
        exc_info=(type(exception), exception, exception.__traceback__),
    )
    config = BaseConfig.global_config()

    if not config.OTEL.IS_ENABLED or not config.OTEL.TRACES_ENABLED:
        return

    try:
        from opentelemetry import trace

        from archipy.helpers.utils.otel_utils import OtelUtils

        span = trace.get_current_span()
        if span.is_recording():
            span.record_exception(exception)
            status = OtelUtils.status_for_exception(exception)
            if status is not None:
                span.set_status(status)
    except ImportError:
        logger.debug("opentelemetry is not installed, cannot record exception on span.")

Validation

Shared formatting and error mapping for validation failures.

archipy.helpers.errors.validation.SupportsErrors

Bases: Protocol

A validation failure that lists its problems (pydantic and FastAPI request validation errors).

Source code in archipy/helpers/errors/validation.py 
18
19
20
21
22
23
class SupportsErrors(Protocol):
    """A validation failure that lists its problems (pydantic and FastAPI request validation errors)."""

    def errors(self) -> Iterable[ErrorDetails]:
        """Return the individual validation problems."""
        ...

archipy.helpers.errors.validation.SupportsErrors.errors 

errors() -> Iterable[ErrorDetails]

Return the individual validation problems.

Source code in archipy/helpers/errors/validation.py 
21
22
23
def errors(self) -> Iterable[ErrorDetails]:
    """Return the individual validation problems."""
    ...

archipy.helpers.errors.validation.format_validation_errors 

format_validation_errors(
    validation_error: SupportsErrors,
    *,
    include_type: bool = False,
    include_input: bool = False,
) -> list[dict[str, str]]

Formats validation errors into a structured format.

Parameters:

Name Type Description Default
validation_error SupportsErrors

The pydantic ValidationError or FastAPI RequestValidationError to format.

 required
include_type bool

Whether to include the error type in the output. Defaults to False.

 False
include_input bool

Whether to echo the rejected input value (as value). Defaults to False, because the input may be a password, token or other secret and must not reach the client or logs.

 False

Returns:

Type Description
list[dict[str, str]]

list[dict[str, str]]: A list of formatted validation error details.
Source code in archipy/helpers/errors/validation.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def format_validation_errors(
    validation_error: SupportsErrors,
    *,
    include_type: bool = False,
    include_input: bool = False,
) -> list[dict[str, str]]:
    """Formats validation errors into a structured format.

    Args:
        validation_error: The pydantic ``ValidationError`` or FastAPI ``RequestValidationError`` to format.
        include_type (bool): Whether to include the error type in the output. Defaults to False.
        include_input (bool): Whether to echo the rejected input value (as ``value``). Defaults to False,
            because the input may be a password, token or other secret and must not reach the client or logs.

    Returns:
        list[dict[str, str]]: A list of formatted validation error details.
    """
    formatted_errors = []
    for error in validation_error.errors():
        error_dict = {
            "field": ".".join(str(x) for x in error["loc"]),
            "message": error["msg"],
        }
        if include_input:
            error_dict["value"] = str(error.get("input", ""))
        if include_type:
            error_dict["type"] = error["type"]
        formatted_errors.append(error_dict)

    return formatted_errors

archipy.helpers.errors.validation.invalid_argument_from_validation 

invalid_argument_from_validation(
    validation_error: ValidationError,
) -> InvalidArgumentError

Build the INVALID_ARGUMENT error that describes a request validation failure.

Parameters:

Name Type Description Default
validation_error ValidationError

The Pydantic validation error.

 required

Returns:

Name Type Description
InvalidArgumentError InvalidArgumentError

The error carrying the (input-free) validation details.
Source code in archipy/helpers/errors/validation.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def invalid_argument_from_validation(validation_error: ValidationError) -> InvalidArgumentError:
    """Build the INVALID_ARGUMENT error that describes a request validation failure.

    Args:
        validation_error: The Pydantic validation error.

    Returns:
        InvalidArgumentError: The error carrying the (input-free) validation details.
    """
    validation_details = format_validation_errors(validation_error, include_type=True)
    return InvalidArgumentError(
        argument_name="request_validation",
        additional_data={"validation_errors": validation_details, "error_count": len(validation_error.errors())},
    )

archipy.helpers.errors.validation.request_validation 

request_validation() -> Generator[None]

Report a Pydantic ValidationError raised inside the block as INVALID_ARGUMENT.

The gRPC exception interceptor already reports every ValidationError as INVALID_ARGUMENT by default. Use this context manager to mark the code that parses the client's request explicitly, which keeps that mapping when the interceptor is created with validation_errors_as_internal=True.

Example 
def Create(self, request, context):
    with request_validation():
        dto = CreateUserDTO(name=request.name, email=request.email)

Raises:

Type Description
InvalidArgumentError

When the block raises pydantic.ValidationError.
Source code in archipy/helpers/errors/validation.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
@contextmanager
def request_validation() -> Generator[None]:
    """Report a Pydantic ``ValidationError`` raised inside the block as INVALID_ARGUMENT.

    The gRPC exception interceptor already reports every ``ValidationError`` as INVALID_ARGUMENT by default. Use this
    context manager to mark the code that parses the *client's request* explicitly, which keeps that mapping when the
    interceptor is created with ``validation_errors_as_internal=True``.

    Example:
        ```python
        def Create(self, request, context):
            with request_validation():
                dto = CreateUserDTO(name=request.name, email=request.email)
        ```

    Raises:
        InvalidArgumentError: When the block raises ``pydantic.ValidationError``.
    """
    try:
        yield
    except ValidationError as validation_error:
        raise invalid_argument_from_validation(validation_error) from validation_error

Mapping

Transport-agnostic mapping from any exception to the BaseError a client should see.

archipy.helpers.errors.mapping.resolve_error 

resolve_error(
    error: Exception,
    *,
    validation_errors_as_internal: bool = False,
) -> BaseError

Return the BaseError to send to the client for error.

A BaseError passes through. A pydantic ValidationError becomes INVALID_ARGUMENT (the gRPC counterpart of HTTP 422), or INTERNAL when validation_errors_as_internal is set. Anything else becomes a generic INTERNAL error, so internals never reach the client.

Parameters:

Name Type Description Default
error Exception

The exception raised by the handler.

 required
validation_errors_as_internal bool

Report a ValidationError as INTERNAL instead of INVALID_ARGUMENT.

 False

Returns:

Name Type Description
BaseError BaseError

The error to report.
Source code in archipy/helpers/errors/mapping.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
def resolve_error(error: Exception, *, validation_errors_as_internal: bool = False) -> BaseError:
    """Return the ``BaseError`` to send to the client for ``error``.

    A ``BaseError`` passes through. A pydantic ``ValidationError`` becomes INVALID_ARGUMENT (the gRPC counterpart of
    HTTP 422), or INTERNAL when ``validation_errors_as_internal`` is set. Anything else becomes a generic INTERNAL
    error, so internals never reach the client.

    Args:
        error: The exception raised by the handler.
        validation_errors_as_internal: Report a ``ValidationError`` as INTERNAL instead of INVALID_ARGUMENT.

    Returns:
        BaseError: The error to report.
    """
    if isinstance(error, BaseError):
        return error
    if isinstance(error, ValidationError) and not validation_errors_as_internal:
        return invalid_argument_from_validation(error)
    return InternalError()

FastAPI

FastAPI exception handlers that turn errors into JSON responses.

archipy.helpers.errors.fastapi.handlers.create_error_response 

create_error_response(
    exception: BaseError, *, capture: bool = True
) -> JSONResponse

Creates a standardized error response.

Parameters:

Name Type Description Default
exception BaseError

The exception to be converted into a response.

 required
capture bool

Whether to log and record exception first. Pass False when the underlying cause was already captured. Defaults to True.

 True

Returns:

Name Type Description
JSONResponse JSONResponse

A JSON response containing the exception details.
Source code in archipy/helpers/errors/fastapi/handlers.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def create_error_response(exception: BaseError, *, capture: bool = True) -> JSONResponse:
    """Creates a standardized error response.

    Args:
        exception (BaseError): The exception to be converted into a response.
        capture (bool): Whether to log and record ``exception`` first. Pass False when the underlying
            cause was already captured. Defaults to True.

    Returns:
        JSONResponse: A JSON response containing the exception details.
    """
    if capture:
        capture_exception(exception)
    status_code = exception.http_status or HTTPStatus.INTERNAL_SERVER_ERROR.value
    return JSONResponse(status_code=status_code, content=exception.to_dict())

archipy.helpers.errors.fastapi.handlers.validation_response 

validation_response(
    exception: ValidationError | RequestValidationError,
) -> JSONResponse

Creates the 422 response for a validation failure.

Parameters:

Name Type Description Default
exception ValidationError | RequestValidationError

The pydantic or FastAPI request validation error.

 required

Returns:

Name Type Description
JSONResponse JSONResponse

A 422 response listing the invalid fields.
Source code in archipy/helpers/errors/fastapi/handlers.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def validation_response(exception: ValidationError | RequestValidationError) -> JSONResponse:
    """Creates the 422 response for a validation failure.

    Args:
        exception: The pydantic or FastAPI request validation error.

    Returns:
        JSONResponse: A 422 response listing the invalid fields.
    """
    capture_exception(exception)
    return JSONResponse(
        status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
        content={
            "error": "VALIDATION_ERROR",
            "detail": format_validation_errors(exception, include_type=isinstance(exception, RequestValidationError)),
        },
    )

archipy.helpers.errors.fastapi.handlers.handle_fastapi_exception async 

handle_fastapi_exception(
    _request: Request, exception: BaseError
) -> JSONResponse

Handles a BaseError and returns its JSON response.

Parameters:

Name Type Description Default
_request Request

The incoming request.

 required
exception BaseError

The exception to handle.

 required

Returns:

Name Type Description
JSONResponse JSONResponse

A JSON response containing the exception details.
Source code in archipy/helpers/errors/fastapi/handlers.py 
56
57
58
59
60
61
62
63
64
65
66
async def handle_fastapi_exception(_request: Request, exception: BaseError) -> JSONResponse:
    """Handles a ``BaseError`` and returns its JSON response.

    Args:
        _request (Request): The incoming request.
        exception (BaseError): The exception to handle.

    Returns:
        JSONResponse: A JSON response containing the exception details.
    """
    return create_error_response(exception)

archipy.helpers.errors.fastapi.handlers.handle_unexpected_exception async 

handle_unexpected_exception(
    _request: Request, exception: Exception
) -> JSONResponse

Handles any other exception; the client only gets a generic error.

Parameters:

Name Type Description Default
_request Request

The incoming request.

 required
exception Exception

The exception to handle.

 required

Returns:

Name Type Description
JSONResponse JSONResponse

A JSON response holding the generic error.
Source code in archipy/helpers/errors/fastapi/handlers.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
async def handle_unexpected_exception(_request: Request, exception: Exception) -> JSONResponse:
    """Handles any other exception; the client only gets a generic error.

    Args:
        _request (Request): The incoming request.
        exception (Exception): The exception to handle.

    Returns:
        JSONResponse: A JSON response holding the generic error.
    """
    # Record the real failure (with its traceback); the client only gets the generic error.
    capture_exception(exception)
    return create_error_response(UnknownError(), capture=False)

archipy.helpers.errors.fastapi.handlers.setup_exception_handlers 

setup_exception_handlers(app: FastAPI) -> None

Configures exception handlers for the FastAPI application.

Parameters:

Name Type Description Default
app FastAPI

The FastAPI application instance.

 required
Source code in archipy/helpers/errors/fastapi/handlers.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def setup_exception_handlers(app: FastAPI) -> None:
    """Configures exception handlers for the FastAPI application.

    Args:
        app (FastAPI): The FastAPI application instance.
    """

    # The wrappers match the (Request, Exception) -> Response signature that FastAPI expects.
    async def validation_wrapper(request: Request, exception: Exception) -> Response:
        if isinstance(exception, ValidationError | RequestValidationError):
            return validation_response(exception)
        return await handle_unexpected_exception(request, exception)

    async def base_error_wrapper(request: Request, exception: Exception) -> Response:
        if isinstance(exception, BaseError):
            return await handle_fastapi_exception(request, exception)
        return await handle_unexpected_exception(request, exception)

    app.add_exception_handler(RequestValidationError, validation_wrapper)
    app.add_exception_handler(ValidationError, validation_wrapper)
    app.add_exception_handler(BaseError, base_error_wrapper)
    app.add_exception_handler(Exception, handle_unexpected_exception)

OpenAPI documentation for the error responses of FastAPI routes.

archipy.helpers.errors.fastapi.openapi.get_fastapi_exception_responses 

get_fastapi_exception_responses(
    exceptions: list[type[BaseError]],
) -> dict[int, dict[str, object]]

Generates OpenAPI response documentation for the given errors.

This method creates OpenAPI-compatible response schemas for FastAPI errors, including validation errors and custom errors.

Parameters:

Name Type Description Default
exceptions list[type[BaseError]]

A list of exception types to generate responses for.

 required

Returns:

Type Description
dict[int, dict[str, object]]

dict[int, dict[str, object]]: A dictionary mapping HTTP status codes to their corresponding response schemas.
Source code in archipy/helpers/errors/fastapi/openapi.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def get_fastapi_exception_responses(exceptions: list[type[BaseError]]) -> dict[int, dict[str, object]]:
    """Generates OpenAPI response documentation for the given errors.

    This method creates OpenAPI-compatible response schemas for FastAPI errors,
    including validation errors and custom errors.

    Args:
        exceptions (list[type[BaseError]]): A list of exception types to generate responses for.

    Returns:
        dict[int, dict[str, object]]: A dictionary mapping HTTP status codes to their corresponding response schemas.
    """
    responses: dict[int, dict[str, object]] = {}

    # Add validation error response by default
    validation_error_response = ValidationErrorResponseDTO()
    if validation_error_response.status_code is not None:
        responses[validation_error_response.status_code] = validation_error_response.model

    exception_schemas = {
        "InvalidPhoneNumberError": {
            "phone_number": {"type": "string", "example": "1234567890", "description": "The invalid phone number"},
        },
        "InvalidLandlineNumberError": {
            "landline_number": {
                "type": "string",
                "example": "02112345678",
                "description": "The invalid landline number",
            },
        },
        "NotFoundError": {
            "resource_type": {
                "type": "string",
                "example": "user",
                "description": "Type of resource that was not found",
            },
        },
        "AlreadyExistsError": {
            "resource_type": {
                "type": "string",
                "example": "user",
                "description": "Type of resource that was not found",
            },
        },
        "InvalidNationalCodeError": {
            "national_code": {
                "type": "string",
                "example": "1234567890",
                "description": "The invalid national code",
            },
        },
        "InvalidArgumentError": {
            "argument": {
                "type": "string",
                "example": "mobile_number",
                "description": "Argument that was invalid",
            },
        },
    }

    for exc in exceptions:
        # Use exception class directly (error details are now class attributes)
        if exc.http_status:
            additional_properties = exception_schemas.get(exc.__name__)
            response = FastAPIErrorResponseDTO(exc, additional_properties)
            if response.status_code is not None:
                responses[response.status_code] = response.model

    return responses

DTOs describing error responses in the OpenAPI documentation.

archipy.helpers.errors.fastapi.response_dtos.FastAPIErrorResponseDTO

Standardized error response model for OpenAPI documentation.

Source code in archipy/helpers/errors/fastapi/response_dtos.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
class FastAPIErrorResponseDTO:
    """Standardized error response model for OpenAPI documentation."""

    def __init__(self, exception: type[BaseError], additional_properties: dict[str, Any] | None = None) -> None:
        """Initialize the error response model.

        Args:
            exception: The exception class (not instance) with error details as class attributes
            additional_properties: Additional properties to include in the response
        """
        self.status_code = exception.http_status

        # Base properties that all errors have
        detail_properties = {
            "code": {"type": "string", "example": exception.code, "description": "Error code identifier"},
            "message_en": {
                "type": "string",
                "example": exception.message_en,
                "description": "Error message in English",
            },
            "message_fa": {
                "type": "string",
                "example": exception.message_fa,
                "description": "Error message in Persian",
            },
            "http_status": {"type": "integer", "example": exception.http_status, "description": "HTTP status code"},
        }

        # Add additional properties if provided
        if additional_properties:
            detail_properties.update(additional_properties)

        self.model: dict[str, object] = {
            "description": exception.message_en,
            "content": {
                "application/json": {
                    "schema": {
                        "type": "object",
                        "properties": {
                            "error": {
                                "type": "string",
                                "example": exception.code,
                                "description": "Error code identifier",
                            },
                            "detail": {
                                "type": "object",
                                "properties": detail_properties,
                                "required": ["code", "message_en", "message_fa", "http_status"],
                                "additionalProperties": False,
                                "description": "Detailed error information",
                            },
                        },
                    },
                },
            },
        }

archipy.helpers.errors.fastapi.response_dtos.FastAPIErrorResponseDTO.status_code instance-attribute 

status_code = exception.http_status

archipy.helpers.errors.fastapi.response_dtos.FastAPIErrorResponseDTO.model instance-attribute 

model: dict[str, object] = {
    "description": exception.message_en,
    "content": {
        "application/json": {
            "schema": {
                "type": "object",
                "properties": {
                    "error": {
                        "type": "string",
                        "example": exception.code,
                        "description": "Error code identifier",
                    },
                    "detail": {
                        "type": "object",
                        "properties": detail_properties,
                        "required": [
                            "code",
                            "message_en",
                            "message_fa",
                            "http_status",
                        ],
                        "additionalProperties": False,
                        "description": "Detailed error information",
                    },
                },
            }
        }
    },
}

archipy.helpers.errors.fastapi.response_dtos.ValidationErrorResponseDTO

Bases: FastAPIErrorResponseDTO

Specific response model for validation errors.

Source code in archipy/helpers/errors/fastapi/response_dtos.py 
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
class ValidationErrorResponseDTO(FastAPIErrorResponseDTO):
    """Specific response model for validation errors."""

    def __init__(self) -> None:
        """Initialize the validation error response model."""
        self.status_code = HTTPStatus.UNPROCESSABLE_ENTITY
        self.model = {
            "description": "Validation Error",
            "content": {
                "application/json": {
                    "schema": {
                        "type": "object",
                        "properties": {
                            "error": {
                                "type": "string",
                                "example": "VALIDATION_ERROR",
                                "description": "Error code identifier",
                            },
                            "detail": {
                                "type": "array",
                                "items": {
                                    "type": "object",
                                    "properties": {
                                        "field": {
                                            "type": "string",
                                            "example": "email",
                                            "description": "Field name that failed validation",
                                        },
                                        "message": {
                                            "type": "string",
                                            "example": "Invalid email format",
                                            "description": "Validation error message",
                                        },
                                    },
                                },
                                "example": [
                                    {"field": "email", "message": "Invalid email format"},
                                    {"field": "password", "message": "Password must be at least 8 characters"},
                                ],
                            },
                        },
                    },
                },
            },
        }

archipy.helpers.errors.fastapi.response_dtos.ValidationErrorResponseDTO.status_code instance-attribute 

status_code = HTTPStatus.UNPROCESSABLE_ENTITY

archipy.helpers.errors.fastapi.response_dtos.ValidationErrorResponseDTO.model instance-attribute 

model = {
    "description": "Validation Error",
    "content": {
        "application/json": {
            "schema": {
                "type": "object",
                "properties": {
                    "error": {
                        "type": "string",
                        "example": "VALIDATION_ERROR",
                        "description": "Error code identifier",
                    },
                    "detail": {
                        "type": "array",
                        "items": {
                            "type": "object",
                            "properties": {
                                "field": {
                                    "type": "string",
                                    "example": "email",
                                    "description": "Field name that failed validation",
                                },
                                "message": {
                                    "type": "string",
                                    "example": "Invalid email format",
                                    "description": "Validation error message",
                                },
                            },
                        },
                        "example": [
                            {
                                "field": "email",
                                "message": "Invalid email format",
                            },
                            {
                                "field": "password",
                                "message": "Password must be at least 8 characters",
                            },
                        ],
                    },
                },
            }
        }
    },
}

gRPC

Abort gRPC calls with the status and message of a BaseError.

archipy.helpers.errors.grpc.abort.grpc_status_code 

grpc_status_code(status: int) -> grpc.StatusCode

Convert an integer status (BaseError.grpc_status) to the gRPC StatusCode member.

Parameters:

Name Type Description Default
status int

The integer gRPC status.

 required

Returns:

Type Description
StatusCode

grpc.StatusCode: The matching member, INTERNAL for an unknown integer.
Source code in archipy/helpers/errors/grpc/abort.py 
24
25
26
27
28
29
30
31
32
33
def grpc_status_code(status: int) -> grpc.StatusCode:
    """Convert an integer status (``BaseError.grpc_status``) to the gRPC ``StatusCode`` member.

    Args:
        status: The integer gRPC status.

    Returns:
        grpc.StatusCode: The matching member, INTERNAL for an unknown integer.
    """
    return _STATUS_CODES.get(status, grpc.StatusCode.INTERNAL)

archipy.helpers.errors.grpc.abort.abort_grpc 

abort_grpc(
    error: BaseError, context: ServicerContext
) -> Awaitable[None]

abort_grpc(
    error: BaseError, context: ServicerContext
) -> None

abort_grpc(
    error: BaseError,
    context: ServicerContext | ServicerContext | None,
) -> Awaitable[None] | None

Abort a gRPC call with the status code and message of error.

The context kind is detected automatically. On a sync context the call is aborted immediately. On an grpc.aio context the abort is returned as an awaitable, so write await abort_grpc(error, context). When error carries additional_data it is sent as the additional_data trailing metadata entry.

Parameters:

Name Type Description Default
error BaseError

The error to report.

 required
context ServicerContext | ServicerContext | None

The sync or async gRPC servicer context to abort.

 required

Returns:

Type Description
Awaitable[None] | None

Awaitable[None] | None: An awaitable for async contexts, None for sync contexts.

Raises:

Type Description
InvalidArgumentError

If the context is None or cannot be aborted.
Source code in archipy/helpers/errors/grpc/abort.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def abort_grpc(
    error: BaseError,
    context: grpc.ServicerContext | AsyncServicerContext | None,
) -> Awaitable[None] | None:
    """Abort a gRPC call with the status code and message of ``error``.

    The context kind is detected automatically. On a sync context the call is aborted immediately. On an
    ``grpc.aio`` context the abort is returned as an awaitable, so write ``await abort_grpc(error, context)``.
    When ``error`` carries ``additional_data`` it is sent as the ``additional_data`` trailing metadata entry.

    Args:
        error: The error to report.
        context: The sync or async gRPC servicer context to abort.

    Returns:
        Awaitable[None] | None: An awaitable for async contexts, None for sync contexts.

    Raises:
        InvalidArgumentError: If the context is None or cannot be aborted.
    """
    if context is None:
        _raise_invalid_context("context_is_none")
    abort = getattr(context, "abort", None)
    if abort is None:
        _raise_invalid_context("missing_abort_method")
    if not callable(abort):
        _raise_invalid_context("abort_not_callable")

    if inspect.iscoroutinefunction(abort):
        return _abort_async(error, context, abort)

    _set_error_metadata(error, context)
    abort(grpc_status_code(error.grpc_status), error.get_message())
    return None

gRPC server interceptor for exception mapping.

archipy.helpers.errors.grpc.interceptors.GrpcExceptionInterceptor

A gRPC server interceptor for centralized exception handling.

Catches the exceptions thrown by gRPC service methods and aborts the call with the matching gRPC status, so service methods need no try/except blocks. BaseError keeps its own status, a pydantic ValidationError is reported as INVALID_ARGUMENT (the counterpart of HTTP 422) and anything else as a generic INTERNAL error.

GrpcExceptionInterceptor() creates the sync interceptor and GrpcExceptionInterceptor(aio=True) the one for grpc.aio servers.

Source code in archipy/helpers/errors/grpc/interceptors.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
class GrpcExceptionInterceptor:
    """A gRPC server interceptor for centralized exception handling.

    Catches the exceptions thrown by gRPC service methods and aborts the call with the matching gRPC status, so
    service methods need no try/except blocks. ``BaseError`` keeps its own status, a pydantic ``ValidationError`` is
    reported as INVALID_ARGUMENT (the counterpart of HTTP 422) and anything else as a generic INTERNAL error.

    ``GrpcExceptionInterceptor()`` creates the sync interceptor and ``GrpcExceptionInterceptor(aio=True)`` the one for
    ``grpc.aio`` servers.
    """

    def __new__(cls, *, aio: bool = False, **_kwargs: object) -> Self:
        """Pick the sync or the aio variant.

        Args:
            aio: Create the interceptor for a ``grpc.aio`` server.
            **_kwargs: Forwarded to ``__init__``.

        Returns:
            Self: The interceptor instance.
        """
        if cls is GrpcExceptionInterceptor:
            variant = _AioGrpcExceptionInterceptor if aio else _SyncGrpcExceptionInterceptor
            return super().__new__(variant)  # ty: ignore[invalid-return-type]
        return super().__new__(cls)

    def __init__(self, *, aio: bool = False, validation_errors_as_internal: bool = False) -> None:  # noqa: ARG002
        """Initialize the interceptor.

        Args:
            aio: Create the interceptor for a ``grpc.aio`` server.
            validation_errors_as_internal: When True, a pydantic ``ValidationError`` is reported as INTERNAL instead
                of INVALID_ARGUMENT (the behavior before 5.15). Defaults to False.
        """
        self._validation_errors_as_internal = validation_errors_as_internal

    def _error_for(self, error: Exception) -> BaseError:
        """Capture ``error`` and return the ``BaseError`` to report."""
        capture_exception(error)
        return resolve_error(error, validation_errors_as_internal=self._validation_errors_as_internal)