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[TRequest, TResult](
        self,
        method: Callable[[TRequest, grpc.ClientCallDetails], TResult],
        request_or_iterator: TRequest,
        call_details: grpc.ClientCallDetails,
    ) -> TResult:
        """Intercepts a gRPC client call.

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

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

    def intercept_unary_unary(
        self,
        continuation: Callable[[grpc.ClientCallDetails, _TRequest], _TResult],
        client_call_details: grpc.ClientCallDetails,
        request: _TRequest,
    ) -> _TResult:
        """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 (_TRequest): The request object.

        Returns:
            _TResult: 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], _TResult],
        client_call_details: grpc.ClientCallDetails,
        request: _TRequest,
    ) -> _TResult:
        """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 (_TRequest): The request object.

        Returns:
            _TResult: 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]], _TResult],
        client_call_details: grpc.ClientCallDetails,
        request_iterator: Iterator[_TRequest],
    ) -> _TResult:
        """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[_TRequest]): The request iterator.

        Returns:
            _TResult: 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]], _TResult],
        client_call_details: grpc.ClientCallDetails,
        request_iterator: Iterator[_TRequest],
    ) -> _TResult:
        """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[_TRequest]): The request iterator.

        Returns:
            _TResult: 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[
        [TRequest, ClientCallDetails], TResult
    ],
    request_or_iterator: TRequest,
    call_details: ClientCallDetails,
) -> TResult

Intercepts a gRPC client call.

Parameters:

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

Returns:

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

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

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

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

    Returns:
        TResult: 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], _TResult
    ],
    client_call_details: ClientCallDetails,
    request: _TRequest,
) -> _TResult

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	_TRequest	The request object.	required

Returns:

Name	Type	Description
_TResult	_TResult	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], _TResult],
    client_call_details: grpc.ClientCallDetails,
    request: _TRequest,
) -> _TResult:
    """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 (_TRequest): The request object.

    Returns:
        _TResult: 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], _TResult
    ],
    client_call_details: ClientCallDetails,
    request: _TRequest,
) -> _TResult

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	_TRequest	The request object.	required

Returns:

Name	Type	Description
_TResult	_TResult	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], _TResult],
    client_call_details: grpc.ClientCallDetails,
    request: _TRequest,
) -> _TResult:
    """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 (_TRequest): The request object.

    Returns:
        _TResult: 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]], _TResult
    ],
    client_call_details: ClientCallDetails,
    request_iterator: Iterator[_TRequest],
) -> _TResult

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[_TRequest]	The request iterator.	required

Returns:

Name	Type	Description
_TResult	_TResult	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]], _TResult],
    client_call_details: grpc.ClientCallDetails,
    request_iterator: Iterator[_TRequest],
) -> _TResult:
    """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[_TRequest]): The request iterator.

    Returns:
        _TResult: 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]], _TResult
    ],
    client_call_details: ClientCallDetails,
    request_iterator: Iterator[_TRequest],
) -> _TResult

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[_TRequest]	The request iterator.	required

Returns:

Name	Type	Description
_TResult	_TResult	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]], _TResult],
    client_call_details: grpc.ClientCallDetails,
    request_iterator: Iterator[_TRequest],
) -> _TResult:
    """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[_TRequest]): The request iterator.

    Returns:
        _TResult: 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[TRequest, TResult](
        self,
        method: Callable[[TRequest, grpc.aio.ClientCallDetails], Awaitable[TResult]],
        request_or_iterator: TRequest,
        call_details: grpc.aio.ClientCallDetails,
    ) -> TResult:
        """Intercepts an asynchronous gRPC client call.

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

        Returns:
            TResult: 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], Awaitable[_TResult]],
        client_call_details: grpc.aio.ClientCallDetails,
        request: _TRequest,
    ) -> _TResult:
        """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 (_TRequest): The request object.

        Returns:
            _TResult: 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], Awaitable[_TResult]],
        client_call_details: grpc.aio.ClientCallDetails,
        request: _TRequest,
    ) -> _TResult:
        """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 (_TRequest): The request object.

        Returns:
            _TResult: 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]],
            Awaitable[_TResult],
        ],
        client_call_details: grpc.aio.ClientCallDetails,
        request_iterator: AsyncIterable[_TRequest] | Iterable[_TRequest],
    ) -> _TResult:
        """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[_TRequest]): The request iterator.

        Returns:
            _TResult: 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]],
            Awaitable[_TResult],
        ],
        client_call_details: grpc.aio.ClientCallDetails,
        request_iterator: AsyncIterable[_TRequest] | Iterable[_TRequest],
    ) -> _TResult:
        """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[_TRequest]): The request iterator.

        Returns:
            _TResult: 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[
        [TRequest, ClientCallDetails], Awaitable[TResult]
    ],
    request_or_iterator: TRequest,
    call_details: ClientCallDetails,
) -> TResult

Intercepts an asynchronous gRPC client call.

Parameters:

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

Returns:

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

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

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

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

    Returns:
        TResult: 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], Awaitable[_TResult]
    ],
    client_call_details: ClientCallDetails,
    request: _TRequest,
) -> _TResult

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	_TRequest	The request object.	required

Returns:

Name	Type	Description
_TResult	_TResult	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], Awaitable[_TResult]],
    client_call_details: grpc.aio.ClientCallDetails,
    request: _TRequest,
) -> _TResult:
    """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 (_TRequest): The request object.

    Returns:
        _TResult: 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], Awaitable[_TResult]
    ],
    client_call_details: ClientCallDetails,
    request: _TRequest,
) -> _TResult

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	_TRequest	The request object.	required

Returns:

Name	Type	Description
_TResult	_TResult	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], Awaitable[_TResult]],
    client_call_details: grpc.aio.ClientCallDetails,
    request: _TRequest,
) -> _TResult:
    """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 (_TRequest): The request object.

    Returns:
        _TResult: 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],
        ],
        Awaitable[_TResult],
    ],
    client_call_details: ClientCallDetails,
    request_iterator: AsyncIterable[_TRequest]
    | Iterable[_TRequest],
) -> _TResult

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[_TRequest]	The request iterator.	required

Returns:

Name	Type	Description
_TResult	_TResult	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]],
        Awaitable[_TResult],
    ],
    client_call_details: grpc.aio.ClientCallDetails,
    request_iterator: AsyncIterable[_TRequest] | Iterable[_TRequest],
) -> _TResult:
    """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[_TRequest]): The request iterator.

    Returns:
        _TResult: 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],
        ],
        Awaitable[_TResult],
    ],
    client_call_details: ClientCallDetails,
    request_iterator: AsyncIterable[_TRequest]
    | Iterable[_TRequest],
) -> _TResult

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[_TRequest]	The request iterator.	required

Returns:

Name	Type	Description
_TResult	_TResult	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]],
        Awaitable[_TResult],
    ],
    client_call_details: grpc.aio.ClientCallDetails,
    request_iterator: AsyncIterable[_TRequest] | Iterable[_TRequest],
) -> _TResult:
    """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[_TRequest]): The request iterator.

    Returns:
        _TResult: 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[..., Any],
        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)

        # ``wraps`` carries decorator metadata (e.g. rate-limit windows) to interceptors placed after this one.
        @functools.wraps(next_handler_method)
        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[..., Any],
    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[..., Any],
    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)

    # ``wraps`` carries decorator metadata (e.g. rate-limit windows) to interceptors placed after this one.
    @functools.wraps(next_handler_method)
    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[..., Any],
        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_stream(
        self,
        method: Callable[..., Any],
        request: object,
        context: grpc.aio.ServicerContext,
        method_name_model: MethodName,
    ) -> AsyncIterator[object]:
        """Intercepts an async gRPC call whose response is a stream.

        Response-streaming handlers must stay async generators, so they cannot go through
        the coroutine-based ``intercept``. The default runs ``intercept`` around the handler
        *invocation* (enough for pre-call checks) and then relays the produced stream.
        Override to wrap the iteration itself (e.g. for error handling or metrics).

        Args:
            method (Callable): The streaming method to be intercepted.
            request (object): The request object (or async iterator for stream requests).
            context (grpc.aio.ServicerContext): The context of the RPC call.
            method_name_model (MethodName): The parsed method name.

        Yields:
            object: Each response message.
        """

        @functools.wraps(method)
        async def start_stream(req: object, ctx: grpc.aio.ServicerContext) -> object:
            return method(req, ctx)

        result = await self.intercept(start_stream, request, context, method_name_model)
        async for item in iterate_stream_result(result):
            yield item

    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)

        # ``wraps`` carries decorator metadata (e.g. rate-limit windows) to interceptors placed after this one.
        @functools.wraps(next_handler_method)
        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)

        @functools.wraps(next_handler_method)
        async def invoke_intercept_stream(
            request: object,
            context: grpc.aio.ServicerContext,
        ) -> AsyncIterator[object]:
            """Invokes the intercepted async streaming method as an async generator.

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

            Yields:
                object: Each response message.
            """
            method_name_model = parse_method_name(handler_call_details.method)
            async for item in self.intercept_stream(next_handler_method, request, context, method_name_model):
                yield item

        return handler_factory(
            invoke_intercept_stream if next_handler.response_streaming else 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[..., Any],
    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[..., Any],
    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_stream async

intercept_stream(
    method: Callable[..., Any],
    request: object,
    context: ServicerContext,
    method_name_model: MethodName,
) -> AsyncIterator[object]

Intercepts an async gRPC call whose response is a stream.

Response-streaming handlers must stay async generators, so they cannot go through the coroutine-based intercept. The default runs intercept around the handler invocation (enough for pre-call checks) and then relays the produced stream. Override to wrap the iteration itself (e.g. for error handling or metrics).

Parameters:

Name	Type	Description	Default
method	Callable	The streaming method to be intercepted.	required
request	object	The request object (or async iterator for stream requests).	required
context	ServicerContext	The context of the RPC call.	required
method_name_model	MethodName	The parsed method name.	required

Yields:

Name	Type	Description
object	AsyncIterator[object]	Each response message.

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

async def intercept_stream(
    self,
    method: Callable[..., Any],
    request: object,
    context: grpc.aio.ServicerContext,
    method_name_model: MethodName,
) -> AsyncIterator[object]:
    """Intercepts an async gRPC call whose response is a stream.

    Response-streaming handlers must stay async generators, so they cannot go through
    the coroutine-based ``intercept``. The default runs ``intercept`` around the handler
    *invocation* (enough for pre-call checks) and then relays the produced stream.
    Override to wrap the iteration itself (e.g. for error handling or metrics).

    Args:
        method (Callable): The streaming method to be intercepted.
        request (object): The request object (or async iterator for stream requests).
        context (grpc.aio.ServicerContext): The context of the RPC call.
        method_name_model (MethodName): The parsed method name.

    Yields:
        object: Each response message.
    """

    @functools.wraps(method)
    async def start_stream(req: object, ctx: grpc.aio.ServicerContext) -> object:
        return method(req, ctx)

    result = await self.intercept(start_stream, request, context, method_name_model)
    async for item in iterate_stream_result(result):
        yield item

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)

    # ``wraps`` carries decorator metadata (e.g. rate-limit windows) to interceptors placed after this one.
    @functools.wraps(next_handler_method)
    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)

    @functools.wraps(next_handler_method)
    async def invoke_intercept_stream(
        request: object,
        context: grpc.aio.ServicerContext,
    ) -> AsyncIterator[object]:
        """Invokes the intercepted async streaming method as an async generator.

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

        Yields:
            object: Each response message.
        """
        method_name_model = parse_method_name(handler_call_details.method)
        async for item in self.intercept_stream(next_handler_method, request, context, method_name_model):
            yield item

    return handler_factory(
        invoke_intercept_stream if next_handler.response_streaming else 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.iterate_stream_result async

iterate_stream_result(
    result: object,
) -> AsyncIterator[object]

Iterate the outcome of invoking a response-streaming async gRPC handler.

Handlers are either async generators (yield responses) or coroutines that push responses through context.write and return None.

Parameters:

Name	Type	Description	Default
result	object	The value returned by calling the handler.	required

Yields:

Name	Type	Description
object	AsyncIterator[object]	Each response message produced by the handler.

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

async def iterate_stream_result(result: object) -> AsyncIterator[object]:
    """Iterate the outcome of invoking a response-streaming async gRPC handler.

    Handlers are either async generators (yield responses) or coroutines that push
    responses through ``context.write`` and return ``None``.

    Args:
        result: The value returned by calling the handler.

    Yields:
        object: Each response message produced by the handler.
    """
    if inspect.isawaitable(result) and not hasattr(result, "__aiter__"):
        result = await result
    if result is None:
        return
    async for item in cast("AsyncIterator[object]", result):
        yield item

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.partition("/")
    *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

The gRPC exception interceptor lives in archipy.helpers.errors.grpc; see Errors. 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.logger module-attribute

logger = logging.getLogger(__name__)

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,
        trusted_proxy_hosts: frozenset[str] | None = None,
        store: RateLimitStore | None = None,
        store_errors: tuple[type[BaseException], ...] = (),
    ) -> None:
        """Initialize the sync gRPC rate-limit interceptor.

        Args:
            rate_limit_config: Rate-limit settings. When None, ``GRPC_RATE_LIMIT`` from global config is used.
            key_prefix: Override for ``GRPC_RATE_LIMIT.KEY_PREFIX``.
            fail_closed: Override for ``GRPC_RATE_LIMIT.FAIL_CLOSED``.
            skip_methods: Full gRPC method names excluded from rate limiting.
            identifier_fn: Custom identity resolver for an RPC.
            identity_from_access_token: Override for ``GRPC_RATE_LIMIT.IDENTITY_FROM_ACCESS_TOKEN``.
            trusted_proxy_hosts: Peer hosts whose ``x-forwarded-for`` header is trusted.
            store: Counter store. When None, a ``RedisAdapter`` built from the global config is used.
            store_errors: Exception types of ``store`` that mean the backend is unavailable, besides the connection,
                OS and timeout errors every store may raise. They follow the ``fail_closed`` setting.
        """
        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,
            trusted_proxy_hosts=trusted_proxy_hosts,
        )
        self._identifier_fn = identifier_fn
        if store is None:
            store, default_errors = self._create_default_store()
            store_errors = (*default_errors, *store_errors)
        self._store: RateLimitStore = store
        self._unavailable_errors = (ConnectionError, OSError, TimeoutError, *store_errors)

    @staticmethod
    def _create_default_store() -> tuple[RateLimitStore, tuple[type[BaseException], ...]]:
        """Build the default Redis store and the Redis errors that mean it is unavailable."""
        from redis.exceptions import RedisError  # noqa: PLC0415

        from archipy.adapters.redis.adapters import RedisAdapter  # noqa: PLC0415

        return RedisAdapter(), (RedisError,)

    def _check(self, key: str, window: RateLimitWindowDTO) -> tuple[int, int]:
        """Increment the rate-limit counter and return limit state."""
        new_value, applied = self._store.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._store.pttl(key), 0

    def _refund(self, keys: Sequence[str]) -> None:
        """Give back the quota consumed in earlier windows by a call that a later window rejected."""
        for key in keys:
            try:
                self._store.increx(key, byint=-1, lbound=0, saturate=True)
            except self._unavailable_errors:
                logger.debug("Could not refund rate-limit quota for %s", key, exc_info=True)

    def intercept(
        self,
        method: Callable[..., Any],
        request: object,
        context: grpc.ServicerContext,
        method_name_model: MethodName,
    ) -> object:
        """Enforce stacked rate-limit windows before invoking the RPC handler."""
        windows = self._windows_for(method, method_name_model)
        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)
        if identity is None:
            return method(request, context)

        consumed: list[str] = []
        for window in windows:
            key = self._build_rate_limit_key(identity, method_name_model, window)
            try:
                pexpire, _remaining = self._check(key, window)
            except self._unavailable_errors 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._refund(consumed)
                self._abort_rate_limited_sync(context, window, pexpire)
                raise RuntimeError("unreachable")
            consumed.append(key)

        return method(request, context)

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

intercept(
    method: Callable[..., Any],
    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[..., Any],
    request: object,
    context: grpc.ServicerContext,
    method_name_model: MethodName,
) -> object:
    """Enforce stacked rate-limit windows before invoking the RPC handler."""
    windows = self._windows_for(method, method_name_model)
    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)
    if identity is None:
        return method(request, context)

    consumed: list[str] = []
    for window in windows:
        key = self._build_rate_limit_key(identity, method_name_model, window)
        try:
            pexpire, _remaining = self._check(key, window)
        except self._unavailable_errors 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._refund(consumed)
            self._abort_rate_limited_sync(context, window, pexpire)
            raise RuntimeError("unreachable")
        consumed.append(key)

    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)

    # ``wraps`` carries decorator metadata (e.g. rate-limit windows) to interceptors placed after this one.
    @functools.wraps(next_handler_method)
    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,
        trusted_proxy_hosts: frozenset[str] | None = None,
        store: AsyncRateLimitStore | None = None,
        store_errors: tuple[type[BaseException], ...] = (),
    ) -> None:
        """Initialize the async gRPC rate-limit interceptor.

        Takes the same arguments as :class:`GrpcServerRateLimitInterceptor`, except that ``store`` is an
        :class:`AsyncRateLimitStore` and defaults to an ``AsyncRedisAdapter`` built from the global config.
        """
        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,
            trusted_proxy_hosts=trusted_proxy_hosts,
        )
        self._identifier_fn = identifier_fn
        if store is None:
            store, default_errors = self._create_default_store()
            store_errors = (*default_errors, *store_errors)
        self._store: AsyncRateLimitStore = store
        self._unavailable_errors = (ConnectionError, OSError, TimeoutError, *store_errors)

    @staticmethod
    def _create_default_store() -> tuple[AsyncRateLimitStore, tuple[type[BaseException], ...]]:
        """Build the default async Redis store and the Redis errors that mean it is unavailable."""
        from redis.exceptions import RedisError  # noqa: PLC0415

        from archipy.adapters.redis.adapters import AsyncRedisAdapter  # noqa: PLC0415

        return AsyncRedisAdapter(), (RedisError,)

    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._store.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._store.pttl(key), 0

    async def _refund(self, keys: Sequence[str]) -> None:
        """Give back the quota consumed in earlier windows by a call that a later window rejected."""
        for key in keys:
            try:
                await self._store.increx(key, byint=-1, lbound=0, saturate=True)
            except self._unavailable_errors:
                logger.debug("Could not refund rate-limit quota for %s", key, exc_info=True)

    async def _enforce_rate_limit_windows_async(
        self,
        context: grpc.aio.ServicerContext,
        method_name_model: MethodName,
        windows: Sequence[RateLimitWindowDTO],
    ) -> None:
        """Enforce rate-limit windows, aborting the call when one is breached.

        Returns once the call may proceed, including when the limiter is fail-open and Redis is unavailable.
        """
        identity = await self._resolve_identity_async(context, method_name_model)
        if identity is None:
            return

        consumed: list[str] = []
        for window in windows:
            key = self._build_rate_limit_key(identity, method_name_model, window)
            try:
                pexpire, _remaining = await self._check(key, window)
            except self._unavailable_errors as exc:
                if self._fail_closed:
                    await self._abort_unavailable_async(context, detail="Rate limiter unavailable")
                    raise RuntimeError("unreachable") from exc
                return

            if pexpire != 0:
                await self._refund(consumed)
                await self._abort_rate_limited_async(context, window, pexpire)
                raise RuntimeError("unreachable")
            consumed.append(key)

    async def intercept(
        self,
        method: Callable[..., Any],
        request: object,
        context: grpc.aio.ServicerContext,
        method_name_model: MethodName,
    ) -> object:
        """Enforce stacked rate-limit windows before invoking the async RPC handler."""
        windows = self._windows_for(method, method_name_model)
        full_method = self._full_method_name(method_name_model)
        if windows and full_method not in self._skip_methods:
            await self._enforce_rate_limit_windows_async(context, method_name_model, windows)

        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[..., Any],
    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[..., Any],
    request: object,
    context: grpc.aio.ServicerContext,
    method_name_model: MethodName,
) -> object:
    """Enforce stacked rate-limit windows before invoking the async RPC handler."""
    windows = self._windows_for(method, method_name_model)
    full_method = self._full_method_name(method_name_model)
    if windows and full_method not in self._skip_methods:
        await self._enforce_rate_limit_windows_async(context, method_name_model, windows)

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

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

intercept_stream(
    method: Callable[..., Any],
    request: object,
    context: ServicerContext,
    method_name_model: MethodName,
) -> AsyncIterator[object]

Intercepts an async gRPC call whose response is a stream.

Response-streaming handlers must stay async generators, so they cannot go through the coroutine-based intercept. The default runs intercept around the handler invocation (enough for pre-call checks) and then relays the produced stream. Override to wrap the iteration itself (e.g. for error handling or metrics).

Parameters:

Name	Type	Description	Default
method	Callable	The streaming method to be intercepted.	required
request	object	The request object (or async iterator for stream requests).	required
context	ServicerContext	The context of the RPC call.	required
method_name_model	MethodName	The parsed method name.	required

Yields:

Name	Type	Description
object	AsyncIterator[object]	Each response message.

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

async def intercept_stream(
    self,
    method: Callable[..., Any],
    request: object,
    context: grpc.aio.ServicerContext,
    method_name_model: MethodName,
) -> AsyncIterator[object]:
    """Intercepts an async gRPC call whose response is a stream.

    Response-streaming handlers must stay async generators, so they cannot go through
    the coroutine-based ``intercept``. The default runs ``intercept`` around the handler
    *invocation* (enough for pre-call checks) and then relays the produced stream.
    Override to wrap the iteration itself (e.g. for error handling or metrics).

    Args:
        method (Callable): The streaming method to be intercepted.
        request (object): The request object (or async iterator for stream requests).
        context (grpc.aio.ServicerContext): The context of the RPC call.
        method_name_model (MethodName): The parsed method name.

    Yields:
        object: Each response message.
    """

    @functools.wraps(method)
    async def start_stream(req: object, ctx: grpc.aio.ServicerContext) -> object:
        return method(req, ctx)

    result = await self.intercept(start_stream, request, context, method_name_model)
    async for item in iterate_stream_result(result):
        yield item

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)

    # ``wraps`` carries decorator metadata (e.g. rate-limit windows) to interceptors placed after this one.
    @functools.wraps(next_handler_method)
    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)

    @functools.wraps(next_handler_method)
    async def invoke_intercept_stream(
        request: object,
        context: grpc.aio.ServicerContext,
    ) -> AsyncIterator[object]:
        """Invokes the intercepted async streaming method as an async generator.

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

        Yields:
            object: Each response message.
        """
        method_name_model = parse_method_name(handler_call_details.method)
        async for item in self.intercept_stream(next_handler_method, request, context, method_name_model):
            yield item

    return handler_factory(
        invoke_intercept_stream if next_handler.response_streaming else 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[object],
) -> dict[str, str]

Normalize gRPC invocation metadata to a string dict.

Parameters:

Name	Type	Description	Default
metadata_items	Iterable[object]	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[object]) -> 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[object],
) -> str | None

Extract a Bearer token from gRPC invocation metadata.

Parameters:

Name	Type	Description	Default
metadata_items	Iterable[object]	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[object]) -> 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[object],
    *,
    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[object]	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[object],
    *,
    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

    try:
        payload = _jwt_utils().verify_access_token(token, auth_config=auth_config or _resolve_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