Interceptors

The helpers/interceptors subpackage provides gRPC server interceptors for exception handling and rate limiting. FastAPI metrics middleware and gRPC metric/trace interceptors were removed in 5.0.0 — use OpenTelemetry via AppUtils / OtelUtils instead (see Observability).

gRPC

base

Abstract base classes for gRPC client and server interceptors.

Base gRPC client interceptor.

archipy.helpers.interceptors.grpc.base.client_interceptor.ClientCallDetails

Bases: _ClientCallDetailsFields, ClientCallDetails

Describes an RPC to be invoked.

This class extends grpc.ClientCallDetails and provides additional fields for RPC details. See https://grpc.github.io/grpc/python/grpc.html#grpc.ClientCallDetails

Source code in archipy/helpers/interceptors/grpc/base/client_interceptor.py

class ClientCallDetails(_ClientCallDetailsFields, grpc.ClientCallDetails):
    """Describes an RPC to be invoked.

    This class extends `grpc.ClientCallDetails` and provides additional fields for RPC details.
    See https://grpc.github.io/grpc/python/grpc.html#grpc.ClientCallDetails
    """

archipy.helpers.interceptors.grpc.base.client_interceptor.ClientCallDetails.method instance-attribute

method: str

archipy.helpers.interceptors.grpc.base.client_interceptor.ClientCallDetails.timeout instance-attribute

timeout: float | None

archipy.helpers.interceptors.grpc.base.client_interceptor.ClientCallDetails.metadata instance-attribute

metadata: Sequence[tuple[str, str | bytes]] | None

archipy.helpers.interceptors.grpc.base.client_interceptor.ClientCallDetails.credentials instance-attribute

credentials: CallCredentials | None

archipy.helpers.interceptors.grpc.base.client_interceptor.ClientCallDetails.wait_for_ready instance-attribute

wait_for_ready: bool | None

archipy.helpers.interceptors.grpc.base.client_interceptor.ClientCallDetails.compression instance-attribute

compression: Compression | None

archipy.helpers.interceptors.grpc.base.client_interceptor.BaseGrpcClientInterceptor

Bases: UnaryUnaryClientInterceptor, UnaryStreamClientInterceptor, StreamUnaryClientInterceptor, StreamStreamClientInterceptor

Base class for gRPC client interceptors.

This class provides a base implementation for intercepting gRPC client calls. It supports unary-unary, unary-stream, stream-unary, and stream-stream RPCs.

Source code in archipy/helpers/interceptors/grpc/base/client_interceptor.py

class BaseGrpcClientInterceptor(
    grpc.UnaryUnaryClientInterceptor,
    grpc.UnaryStreamClientInterceptor,
    grpc.StreamUnaryClientInterceptor,
    grpc.StreamStreamClientInterceptor,
    metaclass=abc.ABCMeta,
):
    """Base class for gRPC client interceptors.

    This class provides a base implementation for intercepting gRPC client calls.
    It supports unary-unary, unary-stream, stream-unary, and stream-stream RPCs.
    """

    @abc.abstractmethod
    def intercept(
        self,
        method: Callable,
        request_or_iterator: Any,
        call_details: grpc.ClientCallDetails,
    ) -> Any:
        """Intercepts a gRPC client call.

        Args:
            method (Callable): The continuation function to call.
            request_or_iterator (Any): The request or request iterator.
            call_details (grpc.ClientCallDetails): Details of the RPC call.

        Returns:
            Any: The result of the intercepted RPC call.
        """
        return method(request_or_iterator, call_details)

    def intercept_unary_unary(
        self,
        continuation: Callable[[grpc.ClientCallDetails, _TRequest], Any],
        client_call_details: grpc.ClientCallDetails,
        request: _TRequest,
    ) -> Any:
        """Intercepts a unary-unary RPC call.

        Args:
            continuation (Callable): The continuation function to call.
            client_call_details (grpc.ClientCallDetails): Details of the RPC call.
            request (Any): The request object.

        Returns:
            Any: The result of the intercepted RPC call.
        """
        return self.intercept(_swap_args(continuation), request, client_call_details)

    def intercept_unary_stream(
        self,
        continuation: Callable[[grpc.ClientCallDetails, _TRequest], Any],
        client_call_details: grpc.ClientCallDetails,
        request: _TRequest,
    ) -> Any:
        """Intercepts a unary-stream RPC call.

        Args:
            continuation (Callable): The continuation function to call.
            client_call_details (grpc.ClientCallDetails): Details of the RPC call.
            request (Any): The request object.

        Returns:
            Any: The result of the intercepted RPC call.
        """
        return self.intercept(_swap_args(continuation), request, client_call_details)

    def intercept_stream_unary(
        self,
        continuation: Callable[[grpc.ClientCallDetails, Iterator[_TRequest]], Any],
        client_call_details: grpc.ClientCallDetails,
        request_iterator: Iterator[_TRequest],
    ) -> Any:
        """Intercepts a stream-unary RPC call.

        Args:
            continuation (Callable): The continuation function to call.
            client_call_details (grpc.ClientCallDetails): Details of the RPC call.
            request_iterator (Iterator[Any]): The request iterator.

        Returns:
            Any: The result of the intercepted RPC call.
        """
        return self.intercept(_swap_args(continuation), request_iterator, client_call_details)

    def intercept_stream_stream(
        self,
        continuation: Callable[[grpc.ClientCallDetails, Iterator[_TRequest]], Any],
        client_call_details: grpc.ClientCallDetails,
        request_iterator: Iterator[_TRequest],
    ) -> Any:
        """Intercepts a stream-stream RPC call.

        Args:
            continuation (Callable): The continuation function to call.
            client_call_details (grpc.ClientCallDetails): Details of the RPC call.
            request_iterator (Iterator[Any]): The request iterator.

        Returns:
            Any: The result of the intercepted RPC call.
        """
        return self.intercept(_swap_args(continuation), request_iterator, client_call_details)

archipy.helpers.interceptors.grpc.base.client_interceptor.BaseGrpcClientInterceptor.intercept abstractmethod

intercept(
    method: Callable,
    request_or_iterator: Any,
    call_details: ClientCallDetails,
) -> Any

Intercepts a gRPC client call.

Parameters:

Name	Type	Description	Default
method	Callable	The continuation function to call.	required
request_or_iterator	Any	The request or request iterator.	required
call_details	ClientCallDetails	Details of the RPC call.	required

Returns:

Name	Type	Description
Any	Any	The result of the intercepted RPC call.

Source code in archipy/helpers/interceptors/grpc/base/client_interceptor.py

@abc.abstractmethod
def intercept(
    self,
    method: Callable,
    request_or_iterator: Any,
    call_details: grpc.ClientCallDetails,
) -> Any:
    """Intercepts a gRPC client call.

    Args:
        method (Callable): The continuation function to call.
        request_or_iterator (Any): The request or request iterator.
        call_details (grpc.ClientCallDetails): Details of the RPC call.

    Returns:
        Any: The result of the intercepted RPC call.
    """
    return method(request_or_iterator, call_details)

archipy.helpers.interceptors.grpc.base.client_interceptor.BaseGrpcClientInterceptor.intercept_unary_unary

intercept_unary_unary(
    continuation: Callable[
        [ClientCallDetails, _TRequest], Any
    ],
    client_call_details: ClientCallDetails,
    request: _TRequest,
) -> Any

Intercepts a unary-unary RPC call.

Parameters:

Name	Type	Description	Default
continuation	Callable	The continuation function to call.	required
client_call_details	ClientCallDetails	Details of the RPC call.	required
request	Any	The request object.	required

Returns:

Name	Type	Description
Any	Any	The result of the intercepted RPC call.

Source code in archipy/helpers/interceptors/grpc/base/client_interceptor.py

def intercept_unary_unary(
    self,
    continuation: Callable[[grpc.ClientCallDetails, _TRequest], Any],
    client_call_details: grpc.ClientCallDetails,
    request: _TRequest,
) -> Any:
    """Intercepts a unary-unary RPC call.

    Args:
        continuation (Callable): The continuation function to call.
        client_call_details (grpc.ClientCallDetails): Details of the RPC call.
        request (Any): The request object.

    Returns:
        Any: The result of the intercepted RPC call.
    """
    return self.intercept(_swap_args(continuation), request, client_call_details)

archipy.helpers.interceptors.grpc.base.client_interceptor.BaseGrpcClientInterceptor.intercept_unary_stream

intercept_unary_stream(
    continuation: Callable[
        [ClientCallDetails, _TRequest], Any
    ],
    client_call_details: ClientCallDetails,
    request: _TRequest,
) -> Any

Intercepts a unary-stream RPC call.

Parameters:

Name	Type	Description	Default
continuation	Callable	The continuation function to call.	required
client_call_details	ClientCallDetails	Details of the RPC call.	required
request	Any	The request object.	required

Returns:

Name	Type	Description
Any	Any	The result of the intercepted RPC call.

Source code in archipy/helpers/interceptors/grpc/base/client_interceptor.py

def intercept_unary_stream(
    self,
    continuation: Callable[[grpc.ClientCallDetails, _TRequest], Any],
    client_call_details: grpc.ClientCallDetails,
    request: _TRequest,
) -> Any:
    """Intercepts a unary-stream RPC call.

    Args:
        continuation (Callable): The continuation function to call.
        client_call_details (grpc.ClientCallDetails): Details of the RPC call.
        request (Any): The request object.

    Returns:
        Any: The result of the intercepted RPC call.
    """
    return self.intercept(_swap_args(continuation), request, client_call_details)

archipy.helpers.interceptors.grpc.base.client_interceptor.BaseGrpcClientInterceptor.intercept_stream_unary

intercept_stream_unary(
    continuation: Callable[
        [ClientCallDetails, Iterator[_TRequest]], Any
    ],
    client_call_details: ClientCallDetails,
    request_iterator: Iterator[_TRequest],
) -> Any

Intercepts a stream-unary RPC call.

Parameters:

Name	Type	Description	Default
continuation	Callable	The continuation function to call.	required
client_call_details	ClientCallDetails	Details of the RPC call.	required
request_iterator	Iterator[Any]	The request iterator.	required

Returns:

Name	Type	Description
Any	Any	The result of the intercepted RPC call.

Source code in archipy/helpers/interceptors/grpc/base/client_interceptor.py

def intercept_stream_unary(
    self,
    continuation: Callable[[grpc.ClientCallDetails, Iterator[_TRequest]], Any],
    client_call_details: grpc.ClientCallDetails,
    request_iterator: Iterator[_TRequest],
) -> Any:
    """Intercepts a stream-unary RPC call.

    Args:
        continuation (Callable): The continuation function to call.
        client_call_details (grpc.ClientCallDetails): Details of the RPC call.
        request_iterator (Iterator[Any]): The request iterator.

    Returns:
        Any: The result of the intercepted RPC call.
    """
    return self.intercept(_swap_args(continuation), request_iterator, client_call_details)

archipy.helpers.interceptors.grpc.base.client_interceptor.BaseGrpcClientInterceptor.intercept_stream_stream

intercept_stream_stream(
    continuation: Callable[
        [ClientCallDetails, Iterator[_TRequest]], Any
    ],
    client_call_details: ClientCallDetails,
    request_iterator: Iterator[_TRequest],
) -> Any

Intercepts a stream-stream RPC call.

Parameters:

Name	Type	Description	Default
continuation	Callable	The continuation function to call.	required
client_call_details	ClientCallDetails	Details of the RPC call.	required
request_iterator	Iterator[Any]	The request iterator.	required

Returns:

Name	Type	Description
Any	Any	The result of the intercepted RPC call.

Source code in archipy/helpers/interceptors/grpc/base/client_interceptor.py

def intercept_stream_stream(
    self,
    continuation: Callable[[grpc.ClientCallDetails, Iterator[_TRequest]], Any],
    client_call_details: grpc.ClientCallDetails,
    request_iterator: Iterator[_TRequest],
) -> Any:
    """Intercepts a stream-stream RPC call.

    Args:
        continuation (Callable): The continuation function to call.
        client_call_details (grpc.ClientCallDetails): Details of the RPC call.
        request_iterator (Iterator[Any]): The request iterator.

    Returns:
        Any: The result of the intercepted RPC call.
    """
    return self.intercept(_swap_args(continuation), request_iterator, client_call_details)

archipy.helpers.interceptors.grpc.base.client_interceptor.AsyncClientCallDetails

Bases: _AsyncClientCallDetailsFields, ClientCallDetails

Describes an RPC to be invoked in an asynchronous context.

This class extends grpc.aio.ClientCallDetails and provides additional fields for RPC details. See https://grpc.github.io/grpc/python/grpc.html#grpc.ClientCallDetails

Source code in archipy/helpers/interceptors/grpc/base/client_interceptor.py

class AsyncClientCallDetails(_AsyncClientCallDetailsFields, grpc.aio.ClientCallDetails):
    """Describes an RPC to be invoked in an asynchronous context.

    This class extends `grpc.aio.ClientCallDetails` and provides additional fields for RPC details.
    See https://grpc.github.io/grpc/python/grpc.html#grpc.ClientCallDetails
    """

archipy.helpers.interceptors.grpc.base.client_interceptor.AsyncClientCallDetails.method instance-attribute

method: str

archipy.helpers.interceptors.grpc.base.client_interceptor.AsyncClientCallDetails.timeout instance-attribute

timeout: float | None

archipy.helpers.interceptors.grpc.base.client_interceptor.AsyncClientCallDetails.metadata instance-attribute

metadata: Sequence[tuple[str, str | bytes]] | None

archipy.helpers.interceptors.grpc.base.client_interceptor.AsyncClientCallDetails.credentials instance-attribute

credentials: CallCredentials | None

archipy.helpers.interceptors.grpc.base.client_interceptor.AsyncClientCallDetails.wait_for_ready instance-attribute

wait_for_ready: bool | None

archipy.helpers.interceptors.grpc.base.client_interceptor.BaseAsyncGrpcClientInterceptor

Bases: UnaryUnaryClientInterceptor, UnaryStreamClientInterceptor, StreamUnaryClientInterceptor, StreamStreamClientInterceptor

Base class for asynchronous gRPC client interceptors.

This class provides a base implementation for intercepting asynchronous gRPC client calls. It supports unary-unary, unary-stream, stream-unary, and stream-stream RPCs.

Source code in archipy/helpers/interceptors/grpc/base/client_interceptor.py

class BaseAsyncGrpcClientInterceptor(
    grpc.aio.UnaryUnaryClientInterceptor,
    grpc.aio.UnaryStreamClientInterceptor,
    grpc.aio.StreamUnaryClientInterceptor,
    grpc.aio.StreamStreamClientInterceptor,
    metaclass=abc.ABCMeta,
):
    """Base class for asynchronous gRPC client interceptors.

    This class provides a base implementation for intercepting asynchronous gRPC client calls.
    It supports unary-unary, unary-stream, stream-unary, and stream-stream RPCs.
    """

    @abc.abstractmethod
    async def intercept(
        self,
        method: Callable,
        request_or_iterator: Any,
        call_details: grpc.aio.ClientCallDetails,
    ) -> Any:
        """Intercepts an asynchronous gRPC client call.

        Args:
            method (Callable): The continuation function to call.
            request_or_iterator (Any): The request or request iterator.
            call_details (grpc.aio.ClientCallDetails): Details of the RPC call.

        Returns:
            Any: The result of the intercepted RPC call.
        """
        return await method(request_or_iterator, call_details)

    async def intercept_unary_unary(
        self,
        continuation: Callable[[grpc.aio.ClientCallDetails, _TRequest], Any],
        client_call_details: grpc.aio.ClientCallDetails,
        request: _TRequest,
    ) -> Any:
        """Intercepts an asynchronous unary-unary RPC call.

        Args:
            continuation (Callable): The continuation function to call.
            client_call_details (grpc.aio.ClientCallDetails): Details of the RPC call.
            request (Any): The request object.

        Returns:
            Any: The result of the intercepted RPC call.
        """
        return await self.intercept(_swap_args(continuation), request, client_call_details)

    async def intercept_unary_stream(
        self,
        continuation: Callable[[grpc.aio.ClientCallDetails, _TRequest], Any],
        client_call_details: grpc.aio.ClientCallDetails,
        request: _TRequest,
    ) -> Any:
        """Intercepts an asynchronous unary-stream RPC call.

        Args:
            continuation (Callable): The continuation function to call.
            client_call_details (grpc.aio.ClientCallDetails): Details of the RPC call.
            request (Any): The request object.

        Returns:
            Any: The result of the intercepted RPC call.
        """
        return await self.intercept(_swap_args(continuation), request, client_call_details)

    async def intercept_stream_unary(
        self,
        continuation: Callable[[grpc.aio.ClientCallDetails, AsyncIterable[_TRequest] | Iterable[_TRequest]], Any],
        client_call_details: grpc.aio.ClientCallDetails,
        request_iterator: AsyncIterable[_TRequest] | Iterable[_TRequest],
    ) -> Any:
        """Intercepts an asynchronous stream-unary RPC call.

        Args:
            continuation (Callable): The continuation function to call.
            client_call_details (grpc.aio.ClientCallDetails): Details of the RPC call.
            request_iterator (Iterator[Any]): The request iterator.

        Returns:
            Any: The result of the intercepted RPC call.
        """
        return await self.intercept(_swap_args(continuation), request_iterator, client_call_details)

    async def intercept_stream_stream(
        self,
        continuation: Callable[[grpc.aio.ClientCallDetails, AsyncIterable[_TRequest] | Iterable[_TRequest]], Any],
        client_call_details: grpc.aio.ClientCallDetails,
        request_iterator: AsyncIterable[_TRequest] | Iterable[_TRequest],
    ) -> Any:
        """Intercepts an asynchronous stream-stream RPC call.

        Args:
            continuation (Callable): The continuation function to call.
            client_call_details (grpc.aio.ClientCallDetails): Details of the RPC call.
            request_iterator (Iterator[Any]): The request iterator.

        Returns:
            Any: The result of the intercepted RPC call.
        """
        return await self.intercept(_swap_args(continuation), request_iterator, client_call_details)

archipy.helpers.interceptors.grpc.base.client_interceptor.BaseAsyncGrpcClientInterceptor.intercept abstractmethod async

intercept(
    method: Callable,
    request_or_iterator: Any,
    call_details: ClientCallDetails,
) -> Any

Intercepts an asynchronous gRPC client call.

Parameters:

Name	Type	Description	Default
method	Callable	The continuation function to call.	required
request_or_iterator	Any	The request or request iterator.	required
call_details	ClientCallDetails	Details of the RPC call.	required

Returns:

Name	Type	Description
Any	Any	The result of the intercepted RPC call.

Source code in archipy/helpers/interceptors/grpc/base/client_interceptor.py

@abc.abstractmethod
async def intercept(
    self,
    method: Callable,
    request_or_iterator: Any,
    call_details: grpc.aio.ClientCallDetails,
) -> Any:
    """Intercepts an asynchronous gRPC client call.

    Args:
        method (Callable): The continuation function to call.
        request_or_iterator (Any): The request or request iterator.
        call_details (grpc.aio.ClientCallDetails): Details of the RPC call.

    Returns:
        Any: The result of the intercepted RPC call.
    """
    return await method(request_or_iterator, call_details)

archipy.helpers.interceptors.grpc.base.client_interceptor.BaseAsyncGrpcClientInterceptor.intercept_unary_unary async

intercept_unary_unary(
    continuation: Callable[
        [ClientCallDetails, _TRequest], Any
    ],
    client_call_details: ClientCallDetails,
    request: _TRequest,
) -> Any

Intercepts an asynchronous unary-unary RPC call.

Parameters:

Name	Type	Description	Default
continuation	Callable	The continuation function to call.	required
client_call_details	ClientCallDetails	Details of the RPC call.	required
request	Any	The request object.	required

Returns:

Name	Type	Description
Any	Any	The result of the intercepted RPC call.

Source code in archipy/helpers/interceptors/grpc/base/client_interceptor.py

async def intercept_unary_unary(
    self,
    continuation: Callable[[grpc.aio.ClientCallDetails, _TRequest], Any],
    client_call_details: grpc.aio.ClientCallDetails,
    request: _TRequest,
) -> Any:
    """Intercepts an asynchronous unary-unary RPC call.

    Args:
        continuation (Callable): The continuation function to call.
        client_call_details (grpc.aio.ClientCallDetails): Details of the RPC call.
        request (Any): The request object.

    Returns:
        Any: The result of the intercepted RPC call.
    """
    return await self.intercept(_swap_args(continuation), request, client_call_details)

archipy.helpers.interceptors.grpc.base.client_interceptor.BaseAsyncGrpcClientInterceptor.intercept_unary_stream async

intercept_unary_stream(
    continuation: Callable[
        [ClientCallDetails, _TRequest], Any
    ],
    client_call_details: ClientCallDetails,
    request: _TRequest,
) -> Any

Intercepts an asynchronous unary-stream RPC call.

Parameters:

Name	Type	Description	Default
continuation	Callable	The continuation function to call.	required
client_call_details	ClientCallDetails	Details of the RPC call.	required
request	Any	The request object.	required

Returns:

Name	Type	Description
Any	Any	The result of the intercepted RPC call.

Source code in archipy/helpers/interceptors/grpc/base/client_interceptor.py

async def intercept_unary_stream(
    self,
    continuation: Callable[[grpc.aio.ClientCallDetails, _TRequest], Any],
    client_call_details: grpc.aio.ClientCallDetails,
    request: _TRequest,
) -> Any:
    """Intercepts an asynchronous unary-stream RPC call.

    Args:
        continuation (Callable): The continuation function to call.
        client_call_details (grpc.aio.ClientCallDetails): Details of the RPC call.
        request (Any): The request object.

    Returns:
        Any: The result of the intercepted RPC call.
    """
    return await self.intercept(_swap_args(continuation), request, client_call_details)

archipy.helpers.interceptors.grpc.base.client_interceptor.BaseAsyncGrpcClientInterceptor.intercept_stream_unary async

intercept_stream_unary(
    continuation: Callable[
        [
            ClientCallDetails,
            AsyncIterable[_TRequest] | Iterable[_TRequest],
        ],
        Any,
    ],
    client_call_details: ClientCallDetails,
    request_iterator: AsyncIterable[_TRequest]
    | Iterable[_TRequest],
) -> Any

Intercepts an asynchronous stream-unary RPC call.

Parameters:

Name	Type	Description	Default
continuation	Callable	The continuation function to call.	required
client_call_details	ClientCallDetails	Details of the RPC call.	required
request_iterator	Iterator[Any]	The request iterator.	required

Returns:

Name	Type	Description
Any	Any	The result of the intercepted RPC call.

Source code in archipy/helpers/interceptors/grpc/base/client_interceptor.py

async def intercept_stream_unary(
    self,
    continuation: Callable[[grpc.aio.ClientCallDetails, AsyncIterable[_TRequest] | Iterable[_TRequest]], Any],
    client_call_details: grpc.aio.ClientCallDetails,
    request_iterator: AsyncIterable[_TRequest] | Iterable[_TRequest],
) -> Any:
    """Intercepts an asynchronous stream-unary RPC call.

    Args:
        continuation (Callable): The continuation function to call.
        client_call_details (grpc.aio.ClientCallDetails): Details of the RPC call.
        request_iterator (Iterator[Any]): The request iterator.

    Returns:
        Any: The result of the intercepted RPC call.
    """
    return await self.intercept(_swap_args(continuation), request_iterator, client_call_details)

archipy.helpers.interceptors.grpc.base.client_interceptor.BaseAsyncGrpcClientInterceptor.intercept_stream_stream async

intercept_stream_stream(
    continuation: Callable[
        [
            ClientCallDetails,
            AsyncIterable[_TRequest] | Iterable[_TRequest],
        ],
        Any,
    ],
    client_call_details: ClientCallDetails,
    request_iterator: AsyncIterable[_TRequest]
    | Iterable[_TRequest],
) -> Any

Intercepts an asynchronous stream-stream RPC call.

Parameters:

Name	Type	Description	Default
continuation	Callable	The continuation function to call.	required
client_call_details	ClientCallDetails	Details of the RPC call.	required
request_iterator	Iterator[Any]	The request iterator.	required

Returns:

Name	Type	Description
Any	Any	The result of the intercepted RPC call.

Source code in archipy/helpers/interceptors/grpc/base/client_interceptor.py

async def intercept_stream_stream(
    self,
    continuation: Callable[[grpc.aio.ClientCallDetails, AsyncIterable[_TRequest] | Iterable[_TRequest]], Any],
    client_call_details: grpc.aio.ClientCallDetails,
    request_iterator: AsyncIterable[_TRequest] | Iterable[_TRequest],
) -> Any:
    """Intercepts an asynchronous stream-stream RPC call.

    Args:
        continuation (Callable): The continuation function to call.
        client_call_details (grpc.aio.ClientCallDetails): Details of the RPC call.
        request_iterator (Iterator[Any]): The request iterator.

    Returns:
        Any: The result of the intercepted RPC call.
    """
    return await self.intercept(_swap_args(continuation), request_iterator, client_call_details)

options: show_root_toc_entry: false heading_level: 3

Base gRPC server interceptor.

archipy.helpers.interceptors.grpc.base.server_interceptor.MethodName

Bases: BaseDTO

A data transfer object (DTO) representing the parsed method name of a gRPC call.

Attributes:

Name	Type	Description
full_name	str	The full name of the method, including package, service, and method.
package	str	The package name.
service	str	The service name.
method	str	The method name.

Source code in archipy/helpers/interceptors/grpc/base/server_interceptor.py

class MethodName(BaseDTO):
    """A data transfer object (DTO) representing the parsed method name of a gRPC call.

    Attributes:
        full_name (str): The full name of the method, including package, service, and method.
        package (str): The package name.
        service (str): The service name.
        method (str): The method name.
    """

    full_name: str
    package: str
    service: str
    method: str

archipy.helpers.interceptors.grpc.base.server_interceptor.MethodName.full_name instance-attribute

full_name: str

archipy.helpers.interceptors.grpc.base.server_interceptor.MethodName.package instance-attribute

package: str

archipy.helpers.interceptors.grpc.base.server_interceptor.MethodName.service instance-attribute

service: str

archipy.helpers.interceptors.grpc.base.server_interceptor.MethodName.method instance-attribute

method: str

archipy.helpers.interceptors.grpc.base.server_interceptor.MethodName.model_config class-attribute instance-attribute

model_config = ConfigDict(
    extra="ignore",
    validate_default=True,
    from_attributes=True,
    frozen=True,
    str_strip_whitespace=True,
    arbitrary_types_allowed=True,
)

archipy.helpers.interceptors.grpc.base.server_interceptor.BaseGrpcServerInterceptor

Bases: ServerInterceptor

Base class for gRPC server interceptors.

This class provides a base implementation for intercepting gRPC server calls. It allows custom logic to be injected into the request/response flow.

Source code in archipy/helpers/interceptors/grpc/base/server_interceptor.py

class BaseGrpcServerInterceptor(grpc.ServerInterceptor, metaclass=abc.ABCMeta):
    """Base class for gRPC server interceptors.

    This class provides a base implementation for intercepting gRPC server calls.
    It allows custom logic to be injected into the request/response flow.
    """

    @abc.abstractmethod
    def intercept(
        self,
        method: Callable,
        request: object,
        context: grpc.ServicerContext,
        method_name_model: MethodName,
    ) -> object:
        """Intercepts a gRPC server call.

        Args:
            method (Callable): The method to be intercepted.
            request (object): The request object.
            context (grpc.ServicerContext): The context of the RPC call.
            method_name_model (str): The full method name (e.g., "/package.Service/Method").

        Returns:
            object: The result of the intercepted method.
        """
        return method(request, context)

    def intercept_service(
        self,
        continuation: Callable[[grpc.HandlerCallDetails], grpc.RpcMethodHandler | None],
        handler_call_details: grpc.HandlerCallDetails,
    ) -> grpc.RpcMethodHandler | None:
        """Intercepts the service call and wraps the handler with custom logic.

        Args:
            continuation: The continuation function to call.
            handler_call_details: Details of the handler call.

        Returns:
            grpc.RpcMethodHandler: The wrapped RPC method handler.
        """
        next_handler = continuation(handler_call_details)
        if next_handler is None:
            return None

        handler_factory, next_handler_method = _get_factory_and_method(next_handler)

        def invoke_intercept_method(request: object, context: grpc.ServicerContext) -> object:
            """Invokes the intercepted method.

            Args:
                request (object): The request object.
                context (grpc.ServicerContext): The context of the RPC call.

            Returns:
                object: The result of the intercepted method.
            """
            method_name_model = parse_method_name(handler_call_details.method)
            return self.intercept(next_handler_method, request, context, method_name_model)

        return handler_factory(
            invoke_intercept_method,
            request_deserializer=next_handler.request_deserializer,
            response_serializer=next_handler.response_serializer,
        )

archipy.helpers.interceptors.grpc.base.server_interceptor.BaseGrpcServerInterceptor.intercept abstractmethod

intercept(
    method: Callable,
    request: object,
    context: ServicerContext,
    method_name_model: MethodName,
) -> object

Intercepts a gRPC server call.

Parameters:

Name	Type	Description	Default
method	Callable	The method to be intercepted.	required
request	object	The request object.	required
context	ServicerContext	The context of the RPC call.	required
method_name_model	str	The full method name (e.g., "/package.Service/Method").	required

Returns:

Name	Type	Description
object	object	The result of the intercepted method.

Source code in archipy/helpers/interceptors/grpc/base/server_interceptor.py

@abc.abstractmethod
def intercept(
    self,
    method: Callable,
    request: object,
    context: grpc.ServicerContext,
    method_name_model: MethodName,
) -> object:
    """Intercepts a gRPC server call.

    Args:
        method (Callable): The method to be intercepted.
        request (object): The request object.
        context (grpc.ServicerContext): The context of the RPC call.
        method_name_model (str): The full method name (e.g., "/package.Service/Method").

    Returns:
        object: The result of the intercepted method.
    """
    return method(request, context)

archipy.helpers.interceptors.grpc.base.server_interceptor.BaseGrpcServerInterceptor.intercept_service

intercept_service(
    continuation: Callable[
        [HandlerCallDetails], RpcMethodHandler | None
    ],
    handler_call_details: HandlerCallDetails,
) -> grpc.RpcMethodHandler | None

Intercepts the service call and wraps the handler with custom logic.

Parameters:

Name	Type	Description	Default
continuation	Callable[[HandlerCallDetails], RpcMethodHandler | None]	The continuation function to call.	required
handler_call_details	HandlerCallDetails	Details of the handler call.	required

Returns:

Type	Description
RpcMethodHandler | None	grpc.RpcMethodHandler: The wrapped RPC method handler.

Source code in archipy/helpers/interceptors/grpc/base/server_interceptor.py

def intercept_service(
    self,
    continuation: Callable[[grpc.HandlerCallDetails], grpc.RpcMethodHandler | None],
    handler_call_details: grpc.HandlerCallDetails,
) -> grpc.RpcMethodHandler | None:
    """Intercepts the service call and wraps the handler with custom logic.

    Args:
        continuation: The continuation function to call.
        handler_call_details: Details of the handler call.

    Returns:
        grpc.RpcMethodHandler: The wrapped RPC method handler.
    """
    next_handler = continuation(handler_call_details)
    if next_handler is None:
        return None

    handler_factory, next_handler_method = _get_factory_and_method(next_handler)

    def invoke_intercept_method(request: object, context: grpc.ServicerContext) -> object:
        """Invokes the intercepted method.

        Args:
            request (object): The request object.
            context (grpc.ServicerContext): The context of the RPC call.

        Returns:
            object: The result of the intercepted method.
        """
        method_name_model = parse_method_name(handler_call_details.method)
        return self.intercept(next_handler_method, request, context, method_name_model)

    return handler_factory(
        invoke_intercept_method,
        request_deserializer=next_handler.request_deserializer,
        response_serializer=next_handler.response_serializer,
    )

archipy.helpers.interceptors.grpc.base.server_interceptor.BaseAsyncGrpcServerInterceptor

Bases: ServerInterceptor

Base class for asynchronous gRPC server interceptors.

This class provides a simplified base implementation for intercepting async gRPC server calls. Unlike the synchronous version, async interceptors work differently and don't need the complex handler wrapping logic.

Source code in archipy/helpers/interceptors/grpc/base/server_interceptor.py

class BaseAsyncGrpcServerInterceptor(grpc.aio.ServerInterceptor, metaclass=abc.ABCMeta):
    """Base class for asynchronous gRPC server interceptors.

    This class provides a simplified base implementation for intercepting async gRPC server calls.
    Unlike the synchronous version, async interceptors work differently and don't need the complex
    handler wrapping logic.
    """

    @abc.abstractmethod
    async def intercept(
        self,
        method: Callable,
        request: object,
        context: grpc.aio.ServicerContext,
        method_name_model: MethodName,
    ) -> object:
        """Intercepts an async gRPC server call.

        Args:
            method (Callable): The method to be intercepted.
            request (object): The request object.
            context (grpc.aio.ServicerContext): The context of the RPC call.
            method_name_model (MethodName): The parsed method name containing package, service, and method components.

        Returns:
            object: The result of the intercepted method.
        """
        return await method(request, context)

    async def intercept_service(
        self,
        continuation: Callable[[grpc.HandlerCallDetails], Awaitable[grpc.RpcMethodHandler | None]],
        handler_call_details: grpc.HandlerCallDetails,
    ) -> grpc.RpcMethodHandler | None:
        """Intercepts the service call using the simplified async pattern.

        For async gRPC, we don't need the complex handler wrapping that sync interceptors require.
        Instead, we can use a much simpler pattern where we just await the continuation and
        then wrap the actual method call.

        Args:
            continuation: The continuation function to call.
            handler_call_details: Details of the handler call.

        Returns:
            grpc.RpcMethodHandler: The wrapped RPC method handler.
        """
        next_handler = await continuation(handler_call_details)
        if next_handler is None:
            return None

        handler_factory, next_handler_method = _get_factory_and_method(next_handler)

        async def invoke_intercept_method(request: object, context: grpc.aio.ServicerContext) -> object:
            """Invokes the intercepted async method.

            Args:
                request (object): The request object.
                context (grpc.aio.ServicerContext): The context of the async RPC call.

            Returns:
                object: The result of the intercepted method.
            """
            method_name_model = parse_method_name(handler_call_details.method)
            return await self.intercept(next_handler_method, request, context, method_name_model)

        return handler_factory(
            invoke_intercept_method,
            request_deserializer=getattr(next_handler, "request_deserializer", None),
            response_serializer=getattr(next_handler, "response_serializer", None),
        )

archipy.helpers.interceptors.grpc.base.server_interceptor.BaseAsyncGrpcServerInterceptor.intercept abstractmethod async

intercept(
    method: Callable,
    request: object,
    context: ServicerContext,
    method_name_model: MethodName,
) -> object

Intercepts an async gRPC server call.

Parameters:

Name	Type	Description	Default
method	Callable	The method to be intercepted.	required
request	object	The request object.	required
context	ServicerContext	The context of the RPC call.	required
method_name_model	MethodName	The parsed method name containing package, service, and method components.	required

Returns:

Name	Type	Description
object	object	The result of the intercepted method.

Source code in archipy/helpers/interceptors/grpc/base/server_interceptor.py

@abc.abstractmethod
async def intercept(
    self,
    method: Callable,
    request: object,
    context: grpc.aio.ServicerContext,
    method_name_model: MethodName,
) -> object:
    """Intercepts an async gRPC server call.

    Args:
        method (Callable): The method to be intercepted.
        request (object): The request object.
        context (grpc.aio.ServicerContext): The context of the RPC call.
        method_name_model (MethodName): The parsed method name containing package, service, and method components.

    Returns:
        object: The result of the intercepted method.
    """
    return await method(request, context)

archipy.helpers.interceptors.grpc.base.server_interceptor.BaseAsyncGrpcServerInterceptor.intercept_service async

intercept_service(
    continuation: Callable[
        [HandlerCallDetails],
        Awaitable[RpcMethodHandler | None],
    ],
    handler_call_details: HandlerCallDetails,
) -> grpc.RpcMethodHandler | None

Intercepts the service call using the simplified async pattern.

For async gRPC, we don't need the complex handler wrapping that sync interceptors require. Instead, we can use a much simpler pattern where we just await the continuation and then wrap the actual method call.

Parameters:

Name	Type	Description	Default
continuation	Callable[[HandlerCallDetails], Awaitable[RpcMethodHandler | None]]	The continuation function to call.	required
handler_call_details	HandlerCallDetails	Details of the handler call.	required

Returns:

Type	Description
RpcMethodHandler | None	grpc.RpcMethodHandler: The wrapped RPC method handler.

Source code in archipy/helpers/interceptors/grpc/base/server_interceptor.py

async def intercept_service(
    self,
    continuation: Callable[[grpc.HandlerCallDetails], Awaitable[grpc.RpcMethodHandler | None]],
    handler_call_details: grpc.HandlerCallDetails,
) -> grpc.RpcMethodHandler | None:
    """Intercepts the service call using the simplified async pattern.

    For async gRPC, we don't need the complex handler wrapping that sync interceptors require.
    Instead, we can use a much simpler pattern where we just await the continuation and
    then wrap the actual method call.

    Args:
        continuation: The continuation function to call.
        handler_call_details: Details of the handler call.

    Returns:
        grpc.RpcMethodHandler: The wrapped RPC method handler.
    """
    next_handler = await continuation(handler_call_details)
    if next_handler is None:
        return None

    handler_factory, next_handler_method = _get_factory_and_method(next_handler)

    async def invoke_intercept_method(request: object, context: grpc.aio.ServicerContext) -> object:
        """Invokes the intercepted async method.

        Args:
            request (object): The request object.
            context (grpc.aio.ServicerContext): The context of the async RPC call.

        Returns:
            object: The result of the intercepted method.
        """
        method_name_model = parse_method_name(handler_call_details.method)
        return await self.intercept(next_handler_method, request, context, method_name_model)

    return handler_factory(
        invoke_intercept_method,
        request_deserializer=getattr(next_handler, "request_deserializer", None),
        response_serializer=getattr(next_handler, "response_serializer", None),
    )

archipy.helpers.interceptors.grpc.base.server_interceptor.parse_method_name

parse_method_name(method_name: str) -> MethodName

Parses a gRPC method name into its components.

Parameters:

Name	Type	Description	Default
method_name	str	The full method name (e.g., "/package.service/method").	required

Returns:

Name	Type	Description
MethodName	MethodName	A MethodName object containing the parsed components.

Source code in archipy/helpers/interceptors/grpc/base/server_interceptor.py

def parse_method_name(method_name: str) -> MethodName:
    """Parses a gRPC method name into its components.

    Args:
        method_name (str): The full method name (e.g., "/package.service/method").

    Returns:
        MethodName: A `MethodName` object containing the parsed components.
    """
    method_full_name = method_name.replace("/", "", 1)
    package_and_service, method = method_full_name.split("/")
    *maybe_package, service = package_and_service.rsplit(".", maxsplit=1)
    package = maybe_package[0] if maybe_package else ""
    return MethodName(full_name=method_full_name, package=package, service=service, method=method)

options: show_root_toc_entry: false heading_level: 3

exception

gRPC server interceptor that catches exceptions and converts them to gRPC status codes.

gRPC server interceptor for exception mapping.

archipy.helpers.interceptors.grpc.exception.server_interceptor.GrpcServerExceptionInterceptor

Bases: BaseGrpcServerInterceptor

A sync gRPC server interceptor for centralized exception handling.

This interceptor catches all exceptions thrown by gRPC service methods and converts them to appropriate gRPC errors, eliminating the need for repetitive try-catch blocks in each service method.

Source code in archipy/helpers/interceptors/grpc/exception/server_interceptor.py

class GrpcServerExceptionInterceptor(BaseGrpcServerInterceptor):
    """A sync gRPC server interceptor for centralized exception handling.

    This interceptor catches all exceptions thrown by gRPC service methods and
    converts them to appropriate gRPC errors, eliminating the need for repetitive
    try-catch blocks in each service method.
    """

    def intercept(
        self,
        method: Callable,
        request: object,
        context: grpc.ServicerContext,
        method_name_model: MethodName,
    ) -> object:
        """Intercepts a sync gRPC server call and handles exceptions.

        Args:
            method: The sync gRPC method being intercepted.
            request: The request object passed to the method.
            context: The context of the sync gRPC call.
            method_name_model: The parsed method name containing package, service, and method components.

        Returns:
            object: The result of the intercepted gRPC method.

        Note:
            This method will not return anything if an exception is handled,
            as the exception handling will abort the gRPC context.
        """
        try:
            # Execute the gRPC method
            result = method(request, context)

        except ValidationError as validation_error:
            BaseUtils.capture_exception(validation_error)
            self._handle_validation_error(validation_error, context)
            raise  # This will never be reached, but satisfies MyPy

        except BaseError as base_error:
            BaseUtils.capture_exception(base_error)
            base_error.abort_grpc_sync(context)
            raise  # This will never be reached, but satisfies MyPy

        except Exception as unexpected_error:
            BaseUtils.capture_exception(unexpected_error)
            self._handle_unexpected_error(unexpected_error, context, method_name_model)
            raise  # This will never be reached, but satisfies MyPy
        else:
            return result

    @staticmethod
    def _handle_validation_error(validation_error: ValidationError, context: grpc.ServicerContext) -> None:
        """Handle Pydantic validation errors.

        Args:
            validation_error: The validation error to handle.
            context: The gRPC context to abort.
        """
        # Format validation errors for better debugging
        validation_details = BaseUtils.format_validation_errors(validation_error, include_type=True)

        InvalidArgumentError(
            argument_name="request_validation",
            additional_data={"validation_errors": validation_details, "error_count": len(validation_error.errors())},
        ).abort_grpc_sync(context)

    @staticmethod
    def _handle_unexpected_error(
        error: Exception,
        context: grpc.ServicerContext,
        method_name_model: MethodName,
    ) -> None:
        """Handle unexpected errors by converting them to internal errors.

        Args:
            error: The unexpected error to handle.
            context: The gRPC context to abort.
            method_name_model: The method name information for better error tracking.
        """
        # Capture the exception for monitoring
        InternalError(
            additional_data={
                "original_error": str(error),
                "error_type": type(error).__name__,
                "service": method_name_model.service,
                "method": method_name_model.method,
                "package": method_name_model.package,
            },
        ).abort_grpc_sync(context)

archipy.helpers.interceptors.grpc.exception.server_interceptor.GrpcServerExceptionInterceptor.intercept

intercept(
    method: Callable,
    request: object,
    context: ServicerContext,
    method_name_model: MethodName,
) -> object

Intercepts a sync gRPC server call and handles exceptions.

Parameters:

Name	Type	Description	Default
method	Callable	The sync gRPC method being intercepted.	required
request	object	The request object passed to the method.	required
context	ServicerContext	The context of the sync gRPC call.	required
method_name_model	MethodName	The parsed method name containing package, service, and method components.	required

Returns:

Name	Type	Description
object	object	The result of the intercepted gRPC method.

Note

This method will not return anything if an exception is handled, as the exception handling will abort the gRPC context.

Source code in archipy/helpers/interceptors/grpc/exception/server_interceptor.py

def intercept(
    self,
    method: Callable,
    request: object,
    context: grpc.ServicerContext,
    method_name_model: MethodName,
) -> object:
    """Intercepts a sync gRPC server call and handles exceptions.

    Args:
        method: The sync gRPC method being intercepted.
        request: The request object passed to the method.
        context: The context of the sync gRPC call.
        method_name_model: The parsed method name containing package, service, and method components.

    Returns:
        object: The result of the intercepted gRPC method.

    Note:
        This method will not return anything if an exception is handled,
        as the exception handling will abort the gRPC context.
    """
    try:
        # Execute the gRPC method
        result = method(request, context)

    except ValidationError as validation_error:
        BaseUtils.capture_exception(validation_error)
        self._handle_validation_error(validation_error, context)
        raise  # This will never be reached, but satisfies MyPy

    except BaseError as base_error:
        BaseUtils.capture_exception(base_error)
        base_error.abort_grpc_sync(context)
        raise  # This will never be reached, but satisfies MyPy

    except Exception as unexpected_error:
        BaseUtils.capture_exception(unexpected_error)
        self._handle_unexpected_error(unexpected_error, context, method_name_model)
        raise  # This will never be reached, but satisfies MyPy
    else:
        return result

archipy.helpers.interceptors.grpc.exception.server_interceptor.GrpcServerExceptionInterceptor.intercept_service

intercept_service(
    continuation: Callable[
        [HandlerCallDetails], RpcMethodHandler | None
    ],
    handler_call_details: HandlerCallDetails,
) -> grpc.RpcMethodHandler | None

Intercepts the service call and wraps the handler with custom logic.

Parameters:

Name	Type	Description	Default
continuation	Callable[[HandlerCallDetails], RpcMethodHandler | None]	The continuation function to call.	required
handler_call_details	HandlerCallDetails	Details of the handler call.	required

Returns:

Type	Description
RpcMethodHandler | None	grpc.RpcMethodHandler: The wrapped RPC method handler.

Source code in archipy/helpers/interceptors/grpc/base/server_interceptor.py

def intercept_service(
    self,
    continuation: Callable[[grpc.HandlerCallDetails], grpc.RpcMethodHandler | None],
    handler_call_details: grpc.HandlerCallDetails,
) -> grpc.RpcMethodHandler | None:
    """Intercepts the service call and wraps the handler with custom logic.

    Args:
        continuation: The continuation function to call.
        handler_call_details: Details of the handler call.

    Returns:
        grpc.RpcMethodHandler: The wrapped RPC method handler.
    """
    next_handler = continuation(handler_call_details)
    if next_handler is None:
        return None

    handler_factory, next_handler_method = _get_factory_and_method(next_handler)

    def invoke_intercept_method(request: object, context: grpc.ServicerContext) -> object:
        """Invokes the intercepted method.

        Args:
            request (object): The request object.
            context (grpc.ServicerContext): The context of the RPC call.

        Returns:
            object: The result of the intercepted method.
        """
        method_name_model = parse_method_name(handler_call_details.method)
        return self.intercept(next_handler_method, request, context, method_name_model)

    return handler_factory(
        invoke_intercept_method,
        request_deserializer=next_handler.request_deserializer,
        response_serializer=next_handler.response_serializer,
    )

archipy.helpers.interceptors.grpc.exception.server_interceptor.AsyncGrpcServerExceptionInterceptor

Bases: BaseAsyncGrpcServerInterceptor

An async gRPC server interceptor for centralized exception handling.

This interceptor catches all exceptions thrown by gRPC service methods and converts them to appropriate gRPC errors, eliminating the need for repetitive try-catch blocks in each service method.

Source code in archipy/helpers/interceptors/grpc/exception/server_interceptor.py

class AsyncGrpcServerExceptionInterceptor(BaseAsyncGrpcServerInterceptor):
    """An async gRPC server interceptor for centralized exception handling.

    This interceptor catches all exceptions thrown by gRPC service methods and
    converts them to appropriate gRPC errors, eliminating the need for repetitive
    try-catch blocks in each service method.
    """

    async def intercept(
        self,
        method: Callable,
        request: object,
        context: grpc.aio.ServicerContext,
        method_name_model: MethodName,
    ) -> object:
        """Intercepts an async gRPC server call and handles exceptions.

        Args:
            method: The async gRPC method being intercepted.
            request: The request object passed to the method.
            context: The context of the async gRPC call.
            method_name_model: The parsed method name containing package, service, and method components.

        Returns:
            object: The result of the intercepted gRPC method.

        Note:
            This method will not return anything if an exception is handled,
            as the exception handling will abort the gRPC context.
        """
        try:
            # Execute the gRPC method
            result = await method(request, context)

        except ValidationError as validation_error:
            BaseUtils.capture_exception(validation_error)
            await self._handle_validation_error(validation_error, context)
            raise  # This will never be reached, but satisfies MyPy

        except BaseError as base_error:
            BaseUtils.capture_exception(base_error)
            await base_error.abort_grpc_async(context)
            raise  # This will never be reached, but satisfies MyPy

        except Exception as unexpected_error:
            BaseUtils.capture_exception(unexpected_error)
            await self._handle_unexpected_error(unexpected_error, context, method_name_model)
            raise  # This will never be reached, but satisfies MyPy
        else:
            return result

    @staticmethod
    async def _handle_validation_error(validation_error: ValidationError, context: grpc.aio.ServicerContext) -> None:
        """Handle Pydantic validation errors.

        Args:
            validation_error: The validation error to handle.
            context: The gRPC context to abort.
        """
        # Format validation errors for better debugging
        validation_details = BaseUtils.format_validation_errors(validation_error, include_type=True)

        await InvalidArgumentError(
            argument_name="request_validation",
            additional_data={"validation_errors": validation_details, "error_count": len(validation_error.errors())},
        ).abort_grpc_async(context)

    @staticmethod
    async def _handle_unexpected_error(
        error: Exception,
        context: grpc.aio.ServicerContext,
        method_name_model: MethodName,
    ) -> None:
        """Handle unexpected errors by converting them to internal errors.

        Args:
            error: The unexpected error to handle.
            context: The gRPC context to abort.
            method_name_model: The method name information for better error tracking.
        """
        # Capture the exception for monitoring
        await InternalError(
            additional_data={
                "original_error": str(error),
                "error_type": type(error).__name__,
                "service": method_name_model.service,
                "method": method_name_model.method,
                "package": method_name_model.package,
            },
        ).abort_grpc_async(context)

archipy.helpers.interceptors.grpc.exception.server_interceptor.AsyncGrpcServerExceptionInterceptor.intercept async

intercept(
    method: Callable,
    request: object,
    context: ServicerContext,
    method_name_model: MethodName,
) -> object

Intercepts an async gRPC server call and handles exceptions.

Parameters:

Name	Type	Description	Default
method	Callable	The async gRPC method being intercepted.	required
request	object	The request object passed to the method.	required
context	ServicerContext	The context of the async gRPC call.	required
method_name_model	MethodName	The parsed method name containing package, service, and method components.	required

Returns:

Name	Type	Description
object	object	The result of the intercepted gRPC method.

Note

This method will not return anything if an exception is handled, as the exception handling will abort the gRPC context.

Source code in archipy/helpers/interceptors/grpc/exception/server_interceptor.py

async def intercept(
    self,
    method: Callable,
    request: object,
    context: grpc.aio.ServicerContext,
    method_name_model: MethodName,
) -> object:
    """Intercepts an async gRPC server call and handles exceptions.

    Args:
        method: The async gRPC method being intercepted.
        request: The request object passed to the method.
        context: The context of the async gRPC call.
        method_name_model: The parsed method name containing package, service, and method components.

    Returns:
        object: The result of the intercepted gRPC method.

    Note:
        This method will not return anything if an exception is handled,
        as the exception handling will abort the gRPC context.
    """
    try:
        # Execute the gRPC method
        result = await method(request, context)

    except ValidationError as validation_error:
        BaseUtils.capture_exception(validation_error)
        await self._handle_validation_error(validation_error, context)
        raise  # This will never be reached, but satisfies MyPy

    except BaseError as base_error:
        BaseUtils.capture_exception(base_error)
        await base_error.abort_grpc_async(context)
        raise  # This will never be reached, but satisfies MyPy

    except Exception as unexpected_error:
        BaseUtils.capture_exception(unexpected_error)
        await self._handle_unexpected_error(unexpected_error, context, method_name_model)
        raise  # This will never be reached, but satisfies MyPy
    else:
        return result

archipy.helpers.interceptors.grpc.exception.server_interceptor.AsyncGrpcServerExceptionInterceptor.intercept_service async

intercept_service(
    continuation: Callable[
        [HandlerCallDetails],
        Awaitable[RpcMethodHandler | None],
    ],
    handler_call_details: HandlerCallDetails,
) -> grpc.RpcMethodHandler | None

Intercepts the service call using the simplified async pattern.

For async gRPC, we don't need the complex handler wrapping that sync interceptors require. Instead, we can use a much simpler pattern where we just await the continuation and then wrap the actual method call.

Parameters:

Name	Type	Description	Default
continuation	Callable[[HandlerCallDetails], Awaitable[RpcMethodHandler | None]]	The continuation function to call.	required
handler_call_details	HandlerCallDetails	Details of the handler call.	required

Returns:

Type	Description
RpcMethodHandler | None	grpc.RpcMethodHandler: The wrapped RPC method handler.

Source code in archipy/helpers/interceptors/grpc/base/server_interceptor.py

async def intercept_service(
    self,
    continuation: Callable[[grpc.HandlerCallDetails], Awaitable[grpc.RpcMethodHandler | None]],
    handler_call_details: grpc.HandlerCallDetails,
) -> grpc.RpcMethodHandler | None:
    """Intercepts the service call using the simplified async pattern.

    For async gRPC, we don't need the complex handler wrapping that sync interceptors require.
    Instead, we can use a much simpler pattern where we just await the continuation and
    then wrap the actual method call.

    Args:
        continuation: The continuation function to call.
        handler_call_details: Details of the handler call.

    Returns:
        grpc.RpcMethodHandler: The wrapped RPC method handler.
    """
    next_handler = await continuation(handler_call_details)
    if next_handler is None:
        return None

    handler_factory, next_handler_method = _get_factory_and_method(next_handler)

    async def invoke_intercept_method(request: object, context: grpc.aio.ServicerContext) -> object:
        """Invokes the intercepted async method.

        Args:
            request (object): The request object.
            context (grpc.aio.ServicerContext): The context of the async RPC call.

        Returns:
            object: The result of the intercepted method.
        """
        method_name_model = parse_method_name(handler_call_details.method)
        return await self.intercept(next_handler_method, request, context, method_name_model)

    return handler_factory(
        invoke_intercept_method,
        request_deserializer=getattr(next_handler, "request_deserializer", None),
        response_serializer=getattr(next_handler, "response_serializer", None),
    )

options: show_root_toc_entry: false heading_level: 3

rate_limit

gRPC server interceptors that enforce decorator-declared Redis rate limits on servicer methods.

gRPC server interceptors that enforce decorator-declared Redis rate limits.

archipy.helpers.interceptors.grpc.rate_limit.grpc_rate_limit_interceptor.GrpcServerRateLimitInterceptor

Bases: _GrpcRateLimitInterceptorMixin, BaseGrpcServerInterceptor

Sync gRPC server interceptor for decorator-declared Redis rate limits.

Source code in archipy/helpers/interceptors/grpc/rate_limit/grpc_rate_limit_interceptor.py

class GrpcServerRateLimitInterceptor(_GrpcRateLimitInterceptorMixin, BaseGrpcServerInterceptor):
    """Sync gRPC server interceptor for decorator-declared Redis rate limits."""

    def __init__(
        self,
        *,
        rate_limit_config: GrpcRateLimitConfig | None = None,
        key_prefix: str | None = None,
        fail_closed: bool | None = None,
        skip_methods: frozenset[str] | None = None,
        identifier_fn: Callable[[grpc.ServicerContext, MethodName], str] | None = None,
        identity_from_access_token: bool | None = None,
    ) -> None:
        """Initialize the sync gRPC rate-limit interceptor."""
        super().__init__(
            rate_limit_config=rate_limit_config,
            key_prefix=key_prefix,
            fail_closed=fail_closed,
            skip_methods=skip_methods,
            identity_from_access_token=identity_from_access_token,
        )
        self._identifier_fn = identifier_fn
        self._redis_client: RedisAdapter = self._create_redis_client()

    @staticmethod
    def _create_redis_client() -> RedisAdapter:
        """Lazily initialized sync Redis client for rate limiting."""
        from archipy.adapters.redis.adapters import RedisAdapter  # noqa: PLC0415

        return RedisAdapter()

    def _check(self, key: str, window: RateLimitWindowDTO) -> tuple[int, int]:
        """Increment the rate-limit counter and return limit state."""
        new_value, applied = self._redis_client.increx(
            key,
            byint=1,
            ubound=window.calls_count,
            saturate=True,
            px=window.window_ms,
            enx=True,
        )
        if applied:
            remaining = max(0, window.calls_count - int(new_value))
            return 0, remaining
        return self._redis_client.pttl(key), 0

    def intercept(
        self,
        method: Callable,
        request: object,
        context: grpc.ServicerContext,
        method_name_model: MethodName,
    ) -> object:
        """Enforce stacked rate-limit windows before invoking the RPC handler."""
        from redis.exceptions import RedisError  # noqa: PLC0415

        windows = RateLimitUtils.get_rate_limit_windows_from_callable(method)
        if not windows:
            return method(request, context)

        full_method = self._full_method_name(method_name_model)
        if full_method in self._skip_methods:
            return method(request, context)

        identity = self._resolve_identity(context, method_name_model)
        for window in windows:
            key = self._build_rate_limit_key(identity, method_name_model, window)
            try:
                pexpire, _remaining = self._check(key, window)
            except (ConnectionError, OSError, TimeoutError, RedisError) as exc:
                if self._fail_closed:
                    self._abort_unavailable_sync(context, detail="Rate limiter unavailable")
                    raise RuntimeError("unreachable") from exc
                return method(request, context)

            if pexpire != 0:
                self._abort_rate_limited_sync(context, window, pexpire)
                raise RuntimeError("unreachable")

        return method(request, context)

archipy.helpers.interceptors.grpc.rate_limit.grpc_rate_limit_interceptor.GrpcServerRateLimitInterceptor.intercept

intercept(
    method: Callable,
    request: object,
    context: ServicerContext,
    method_name_model: MethodName,
) -> object

Enforce stacked rate-limit windows before invoking the RPC handler.

Source code in archipy/helpers/interceptors/grpc/rate_limit/grpc_rate_limit_interceptor.py

def intercept(
    self,
    method: Callable,
    request: object,
    context: grpc.ServicerContext,
    method_name_model: MethodName,
) -> object:
    """Enforce stacked rate-limit windows before invoking the RPC handler."""
    from redis.exceptions import RedisError  # noqa: PLC0415

    windows = RateLimitUtils.get_rate_limit_windows_from_callable(method)
    if not windows:
        return method(request, context)

    full_method = self._full_method_name(method_name_model)
    if full_method in self._skip_methods:
        return method(request, context)

    identity = self._resolve_identity(context, method_name_model)
    for window in windows:
        key = self._build_rate_limit_key(identity, method_name_model, window)
        try:
            pexpire, _remaining = self._check(key, window)
        except (ConnectionError, OSError, TimeoutError, RedisError) as exc:
            if self._fail_closed:
                self._abort_unavailable_sync(context, detail="Rate limiter unavailable")
                raise RuntimeError("unreachable") from exc
            return method(request, context)

        if pexpire != 0:
            self._abort_rate_limited_sync(context, window, pexpire)
            raise RuntimeError("unreachable")

    return method(request, context)

archipy.helpers.interceptors.grpc.rate_limit.grpc_rate_limit_interceptor.GrpcServerRateLimitInterceptor.intercept_service

intercept_service(
    continuation: Callable[
        [HandlerCallDetails], RpcMethodHandler | None
    ],
    handler_call_details: HandlerCallDetails,
) -> grpc.RpcMethodHandler | None

Intercepts the service call and wraps the handler with custom logic.

Parameters:

Name	Type	Description	Default
continuation	Callable[[HandlerCallDetails], RpcMethodHandler | None]	The continuation function to call.	required
handler_call_details	HandlerCallDetails	Details of the handler call.	required

Returns:

Type	Description
RpcMethodHandler | None	grpc.RpcMethodHandler: The wrapped RPC method handler.

Source code in archipy/helpers/interceptors/grpc/base/server_interceptor.py

def intercept_service(
    self,
    continuation: Callable[[grpc.HandlerCallDetails], grpc.RpcMethodHandler | None],
    handler_call_details: grpc.HandlerCallDetails,
) -> grpc.RpcMethodHandler | None:
    """Intercepts the service call and wraps the handler with custom logic.

    Args:
        continuation: The continuation function to call.
        handler_call_details: Details of the handler call.

    Returns:
        grpc.RpcMethodHandler: The wrapped RPC method handler.
    """
    next_handler = continuation(handler_call_details)
    if next_handler is None:
        return None

    handler_factory, next_handler_method = _get_factory_and_method(next_handler)

    def invoke_intercept_method(request: object, context: grpc.ServicerContext) -> object:
        """Invokes the intercepted method.

        Args:
            request (object): The request object.
            context (grpc.ServicerContext): The context of the RPC call.

        Returns:
            object: The result of the intercepted method.
        """
        method_name_model = parse_method_name(handler_call_details.method)
        return self.intercept(next_handler_method, request, context, method_name_model)

    return handler_factory(
        invoke_intercept_method,
        request_deserializer=next_handler.request_deserializer,
        response_serializer=next_handler.response_serializer,
    )

archipy.helpers.interceptors.grpc.rate_limit.grpc_rate_limit_interceptor.AsyncGrpcServerRateLimitInterceptor

Bases: _GrpcRateLimitInterceptorMixin, BaseAsyncGrpcServerInterceptor

Async gRPC server interceptor for decorator-declared Redis rate limits.

Source code in archipy/helpers/interceptors/grpc/rate_limit/grpc_rate_limit_interceptor.py

class AsyncGrpcServerRateLimitInterceptor(_GrpcRateLimitInterceptorMixin, BaseAsyncGrpcServerInterceptor):
    """Async gRPC server interceptor for decorator-declared Redis rate limits."""

    def __init__(
        self,
        *,
        rate_limit_config: GrpcRateLimitConfig | None = None,
        key_prefix: str | None = None,
        fail_closed: bool | None = None,
        skip_methods: frozenset[str] | None = None,
        identifier_fn: Callable[[grpc.aio.ServicerContext, MethodName], str] | None = None,
        identity_from_access_token: bool | None = None,
    ) -> None:
        """Initialize the async gRPC rate-limit interceptor."""
        super().__init__(
            rate_limit_config=rate_limit_config,
            key_prefix=key_prefix,
            fail_closed=fail_closed,
            skip_methods=skip_methods,
            identity_from_access_token=identity_from_access_token,
        )
        self._identifier_fn = identifier_fn
        self._redis_client: AsyncRedisAdapter = self._create_redis_client()

    @staticmethod
    def _create_redis_client() -> AsyncRedisAdapter:
        """Lazily initialized async Redis client for rate limiting."""
        from archipy.adapters.redis.adapters import AsyncRedisAdapter  # noqa: PLC0415

        return AsyncRedisAdapter()

    async def _check(self, key: str, window: RateLimitWindowDTO) -> tuple[int, int]:
        """Increment the rate-limit counter and return limit state."""
        new_value, applied = await self._redis_client.increx(
            key,
            byint=1,
            ubound=window.calls_count,
            saturate=True,
            px=window.window_ms,
            enx=True,
        )
        if applied:
            remaining = max(0, window.calls_count - int(new_value))
            return 0, remaining
        return await self._redis_client.pttl(key), 0

    async def _enforce_rate_limit_windows_async(
        self,
        method: Callable,
        request: object,
        context: grpc.aio.ServicerContext,
        method_name_model: MethodName,
        windows: Sequence[RateLimitWindowDTO],
    ) -> object | None:
        """Enforce rate-limit windows; return handler result on fail-open bypass."""
        from redis.exceptions import RedisError  # noqa: PLC0415

        identity = await self._resolve_identity_async(context, method_name_model)
        for window in windows:
            key = self._build_rate_limit_key(identity, method_name_model, window)
            try:
                pexpire, _remaining = await self._check(key, window)
            except (ConnectionError, OSError, TimeoutError, RedisError) as exc:
                if self._fail_closed:
                    await self._abort_unavailable_async(context, detail="Rate limiter unavailable")
                    raise RuntimeError("unreachable") from exc
                return await self._invoke_maybe_async_handler(method, request, context)

            if pexpire != 0:
                await self._abort_rate_limited_async(context, window, pexpire)
                raise RuntimeError("unreachable")
        return None

    async def intercept(
        self,
        method: Callable,
        request: object,
        context: grpc.aio.ServicerContext,
        method_name_model: MethodName,
    ) -> object:
        """Enforce stacked rate-limit windows before invoking the async RPC handler."""
        windows = RateLimitUtils.get_rate_limit_windows_from_callable(method)
        if not windows:
            return await self._invoke_maybe_async_handler(method, request, context)

        full_method = self._full_method_name(method_name_model)
        if full_method in self._skip_methods:
            return await self._invoke_maybe_async_handler(method, request, context)

        bypass_result = await self._enforce_rate_limit_windows_async(
            method,
            request,
            context,
            method_name_model,
            windows,
        )
        if bypass_result is not None:
            return bypass_result

        return await self._invoke_maybe_async_handler(method, request, context)

archipy.helpers.interceptors.grpc.rate_limit.grpc_rate_limit_interceptor.AsyncGrpcServerRateLimitInterceptor.intercept async

intercept(
    method: Callable,
    request: object,
    context: ServicerContext,
    method_name_model: MethodName,
) -> object

Enforce stacked rate-limit windows before invoking the async RPC handler.

Source code in archipy/helpers/interceptors/grpc/rate_limit/grpc_rate_limit_interceptor.py

async def intercept(
    self,
    method: Callable,
    request: object,
    context: grpc.aio.ServicerContext,
    method_name_model: MethodName,
) -> object:
    """Enforce stacked rate-limit windows before invoking the async RPC handler."""
    windows = RateLimitUtils.get_rate_limit_windows_from_callable(method)
    if not windows:
        return await self._invoke_maybe_async_handler(method, request, context)

    full_method = self._full_method_name(method_name_model)
    if full_method in self._skip_methods:
        return await self._invoke_maybe_async_handler(method, request, context)

    bypass_result = await self._enforce_rate_limit_windows_async(
        method,
        request,
        context,
        method_name_model,
        windows,
    )
    if bypass_result is not None:
        return bypass_result

    return await self._invoke_maybe_async_handler(method, request, context)

archipy.helpers.interceptors.grpc.rate_limit.grpc_rate_limit_interceptor.AsyncGrpcServerRateLimitInterceptor.intercept_service async

intercept_service(
    continuation: Callable[
        [HandlerCallDetails],
        Awaitable[RpcMethodHandler | None],
    ],
    handler_call_details: HandlerCallDetails,
) -> grpc.RpcMethodHandler | None

Intercepts the service call using the simplified async pattern.

For async gRPC, we don't need the complex handler wrapping that sync interceptors require. Instead, we can use a much simpler pattern where we just await the continuation and then wrap the actual method call.

Parameters:

Name	Type	Description	Default
continuation	Callable[[HandlerCallDetails], Awaitable[RpcMethodHandler | None]]	The continuation function to call.	required
handler_call_details	HandlerCallDetails	Details of the handler call.	required

Returns:

Type	Description
RpcMethodHandler | None	grpc.RpcMethodHandler: The wrapped RPC method handler.

Source code in archipy/helpers/interceptors/grpc/base/server_interceptor.py

async def intercept_service(
    self,
    continuation: Callable[[grpc.HandlerCallDetails], Awaitable[grpc.RpcMethodHandler | None]],
    handler_call_details: grpc.HandlerCallDetails,
) -> grpc.RpcMethodHandler | None:
    """Intercepts the service call using the simplified async pattern.

    For async gRPC, we don't need the complex handler wrapping that sync interceptors require.
    Instead, we can use a much simpler pattern where we just await the continuation and
    then wrap the actual method call.

    Args:
        continuation: The continuation function to call.
        handler_call_details: Details of the handler call.

    Returns:
        grpc.RpcMethodHandler: The wrapped RPC method handler.
    """
    next_handler = await continuation(handler_call_details)
    if next_handler is None:
        return None

    handler_factory, next_handler_method = _get_factory_and_method(next_handler)

    async def invoke_intercept_method(request: object, context: grpc.aio.ServicerContext) -> object:
        """Invokes the intercepted async method.

        Args:
            request (object): The request object.
            context (grpc.aio.ServicerContext): The context of the async RPC call.

        Returns:
            object: The result of the intercepted method.
        """
        method_name_model = parse_method_name(handler_call_details.method)
        return await self.intercept(next_handler_method, request, context, method_name_model)

    return handler_factory(
        invoke_intercept_method,
        request_deserializer=getattr(next_handler, "request_deserializer", None),
        response_serializer=getattr(next_handler, "response_serializer", None),
    )

options: show_root_toc_entry: false heading_level: 3

Rate-limit identity helpers for gRPC server RPCs.

archipy.helpers.interceptors.grpc.rate_limit.identifiers.invocation_metadata_to_dict

invocation_metadata_to_dict(
    metadata_items: Iterable[Any],
) -> dict[str, str]

Normalize gRPC invocation metadata to a string dict.

Parameters:

Name	Type	Description	Default
metadata_items	Iterable[Any]	Items from ServicerContext.invocation_metadata().	required

Returns:

Type	Description
dict[str, str]	Metadata keys mapped to decoded string values.

Source code in archipy/helpers/interceptors/grpc/rate_limit/identifiers.py

def invocation_metadata_to_dict(metadata_items: Iterable[Any]) -> dict[str, str]:
    """Normalize gRPC invocation metadata to a string dict.

    Args:
        metadata_items: Items from ``ServicerContext.invocation_metadata()``.

    Returns:
        Metadata keys mapped to decoded string values.
    """
    metadata_dict: dict[str, str] = {}
    for item in metadata_items:
        if hasattr(item, "key") and hasattr(item, "value"):
            key, value = item.key, item.value
        elif isinstance(item, tuple) and len(item) >= _METADATA_PAIR_MIN_LEN:
            key, value = item[0], item[1]
        else:
            continue
        if isinstance(value, bytes):
            metadata_dict[str(key)] = value.decode("utf-8", errors="ignore")
        else:
            metadata_dict[str(key)] = str(value)
    return metadata_dict

archipy.helpers.interceptors.grpc.rate_limit.identifiers.extract_bearer_token_from_metadata

extract_bearer_token_from_metadata(
    metadata_items: Iterable[Any],
) -> str | None

Extract a Bearer token from gRPC invocation metadata.

Parameters:

Name	Type	Description	Default
metadata_items	Iterable[Any]	Items from ServicerContext.invocation_metadata().	required

Returns:

Type	Description
str | None	The raw JWT string, or None when metadata is missing or not Bearer.

Source code in archipy/helpers/interceptors/grpc/rate_limit/identifiers.py

def extract_bearer_token_from_metadata(metadata_items: Iterable[Any]) -> str | None:
    """Extract a Bearer token from gRPC invocation metadata.

    Args:
        metadata_items: Items from ``ServicerContext.invocation_metadata()``.

    Returns:
        The raw JWT string, or None when metadata is missing or not Bearer.
    """
    metadata_dict = invocation_metadata_to_dict(metadata_items)
    authorization = metadata_dict.get("authorization") or metadata_dict.get("Authorization")
    if not authorization:
        return None

    scheme, _, credentials = authorization.partition(" ")
    if scheme.lower() != "bearer" or not credentials:
        return None

    return credentials.strip() or None

archipy.helpers.interceptors.grpc.rate_limit.identifiers.resolve_jwt_access_token_sub_from_metadata

resolve_jwt_access_token_sub_from_metadata(
    metadata_items: Iterable[Any],
    *,
    auth_config: AuthConfig | None = None,
) -> str | None

Resolve the user identity from a verified JWT in gRPC invocation metadata.

Parameters:

Name	Type	Description	Default
metadata_items	Iterable[Any]	Items from ServicerContext.invocation_metadata().	required
auth_config	AuthConfig | None	Optional auth configuration override. When None, uses global config.	None

Returns:

Type	Description
str | None	The token sub claim as a string, or None when no valid access token is present.

Source code in archipy/helpers/interceptors/grpc/rate_limit/identifiers.py

def resolve_jwt_access_token_sub_from_metadata(
    metadata_items: Iterable[Any],
    *,
    auth_config: AuthConfig | None = None,
) -> str | None:
    """Resolve the user identity from a verified JWT in gRPC invocation metadata.

    Args:
        metadata_items: Items from ``ServicerContext.invocation_metadata()``.
        auth_config: Optional auth configuration override. When None, uses global config.

    Returns:
        The token ``sub`` claim as a string, or None when no valid access token is present.
    """
    token = extract_bearer_token_from_metadata(metadata_items)
    if not token:
        return None

    from archipy.helpers.utils.jwt_utils import JWTUtils  # noqa: PLC0415

    try:
        payload = JWTUtils.verify_access_token(token, auth_config=auth_config)
    except InvalidTokenError, TokenExpiredError:
        return None

    sub = payload.get("sub")
    if sub is None:
        return None

    return str(sub)

options: show_root_toc_entry: false heading_level: 3