Interceptors

The helpers/interceptors subpackage provides request/response interceptors for FastAPI and gRPC, covering rate limiting, metrics collection, exception handling, and distributed tracing.

FastAPI

metric

FastAPI middleware interceptor for collecting Prometheus metrics on HTTP requests.

archipy.helpers.interceptors.fastapi.metric.interceptor.FastAPIMetricInterceptor

Bases: BaseHTTPMiddleware

A FastAPI interceptor for collecting and reporting metrics using Prometheus.

This interceptor measures the response time of HTTP requests and records it in a Prometheus histogram. It also tracks the number of active requests using a Prometheus gauge. The interceptor captures errors and logs them for monitoring purposes.

Source code in archipy/helpers/interceptors/fastapi/metric/interceptor.py

class FastAPIMetricInterceptor(BaseHTTPMiddleware):
    """A FastAPI interceptor for collecting and reporting metrics using Prometheus.

    This interceptor measures the response time of HTTP requests and records it in a Prometheus histogram.
    It also tracks the number of active requests using a Prometheus gauge.
    The interceptor captures errors and logs them for monitoring purposes.
    """

    ZERO_TO_ONE_SECONDS_BUCKETS: ClassVar[list[float]] = [i / 1000 for i in range(0, 1000, 5)]
    ONE_TO_FIVE_SECONDS_BUCKETS: ClassVar[list[float]] = [i / 100 for i in range(100, 500, 20)]
    FIVE_TO_THIRTY_SECONDS_BUCKETS: ClassVar[list[float]] = [i / 100 for i in range(500, 3000, 50)]
    TOTAL_BUCKETS: ClassVar[list[float]] = (
        ZERO_TO_ONE_SECONDS_BUCKETS + ONE_TO_FIVE_SECONDS_BUCKETS + FIVE_TO_THIRTY_SECONDS_BUCKETS + [float("inf")]
    )

    RESPONSE_TIME_SECONDS: ClassVar[Histogram] = Histogram(
        "fastapi_response_time_seconds",
        "Time spent processing HTTP request",
        labelnames=("method", "status_code", "path_template"),
        buckets=TOTAL_BUCKETS,
    )

    ACTIVE_REQUESTS: ClassVar[Gauge] = Gauge(
        "fastapi_active_requests",
        "Number of active HTTP requests",
        labelnames=("method", "path_template"),
    )

    _path_template_cache: ClassVar[dict[str, str]] = {}

    def __init__(self, app: ASGIApp) -> None:
        """Initialize the FastAPI metric interceptor.

        Args:
            app (ASGIApp): The ASGI application to wrap.
        """
        super().__init__(app)

    async def dispatch(self, request: Request, call_next: Callable[[Request], Awaitable[Response]]) -> Response:
        """Intercept HTTP requests to measure response time and track active requests.

        Args:
            request (Request): The incoming HTTP request.
            call_next (Callable[[Request], Awaitable[Response]]): The next interceptor or endpoint to call.

        Returns:
            Response: The HTTP response from the endpoint.

        Raises:
            Exception: If an exception occurs during request processing, it is captured and re-raised.
        """
        if not BaseConfig.global_config().PROMETHEUS.IS_ENABLED:
            return await call_next(request)

        path_template = self._get_path_template(request)
        method = request.method

        self.ACTIVE_REQUESTS.labels(method=method, path_template=path_template).inc()

        start_time = time.time()
        status_code = 500

        try:
            response = await call_next(request)
            status_code = response.status_code
        except Exception as exception:
            BaseUtils.capture_exception(exception)
            raise
        else:
            return response
        finally:
            duration = time.time() - start_time
            self.RESPONSE_TIME_SECONDS.labels(
                method=method,
                status_code=status_code,
                path_template=path_template,
            ).observe(duration)
            self.ACTIVE_REQUESTS.labels(method=method, path_template=path_template).dec()

    def _get_path_template(self, request: Request) -> str:
        """Extract path template from request by matching against app routes with in-memory caching.

        Args:
            request (Request): The FastAPI request object.

        Returns:
            str: Path template (e.g., /users/{id}) or raw path if no route found.
        """
        path = request.url.path
        method = request.method
        cache_key = f"{method}:{path}"

        if cache_key in self._path_template_cache:
            return self._path_template_cache[cache_key]

        for route in _flatten_routes(request.app.routes):
            match, _ = route.matches(request.scope)
            if match == Match.FULL:
                path_template = route.path
                self._path_template_cache[cache_key] = path_template
                return path_template

        self._path_template_cache[cache_key] = path
        return path

archipy.helpers.interceptors.fastapi.metric.interceptor.FastAPIMetricInterceptor.ZERO_TO_ONE_SECONDS_BUCKETS class-attribute

ZERO_TO_ONE_SECONDS_BUCKETS: list[float] = [
    (i / 1000) for i in (range(0, 1000, 5))
]

archipy.helpers.interceptors.fastapi.metric.interceptor.FastAPIMetricInterceptor.ONE_TO_FIVE_SECONDS_BUCKETS class-attribute

ONE_TO_FIVE_SECONDS_BUCKETS: list[float] = [
    (i / 100) for i in (range(100, 500, 20))
]

archipy.helpers.interceptors.fastapi.metric.interceptor.FastAPIMetricInterceptor.FIVE_TO_THIRTY_SECONDS_BUCKETS class-attribute

FIVE_TO_THIRTY_SECONDS_BUCKETS: list[float] = [
    (i / 100) for i in (range(500, 3000, 50))
]

archipy.helpers.interceptors.fastapi.metric.interceptor.FastAPIMetricInterceptor.TOTAL_BUCKETS class-attribute

TOTAL_BUCKETS: list[float] = (
    ZERO_TO_ONE_SECONDS_BUCKETS
    + ONE_TO_FIVE_SECONDS_BUCKETS
    + FIVE_TO_THIRTY_SECONDS_BUCKETS
    + [float("inf")]
)

archipy.helpers.interceptors.fastapi.metric.interceptor.FastAPIMetricInterceptor.RESPONSE_TIME_SECONDS class-attribute

RESPONSE_TIME_SECONDS: Histogram = Histogram(
    "fastapi_response_time_seconds",
    "Time spent processing HTTP request",
    labelnames=("method", "status_code", "path_template"),
    buckets=TOTAL_BUCKETS,
)

archipy.helpers.interceptors.fastapi.metric.interceptor.FastAPIMetricInterceptor.ACTIVE_REQUESTS class-attribute

ACTIVE_REQUESTS: Gauge = Gauge(
    "fastapi_active_requests",
    "Number of active HTTP requests",
    labelnames=("method", "path_template"),
)

archipy.helpers.interceptors.fastapi.metric.interceptor.FastAPIMetricInterceptor.dispatch async

dispatch(
    request: Request,
    call_next: Callable[[Request], Awaitable[Response]],
) -> Response

Intercept HTTP requests to measure response time and track active requests.

Parameters:

Name	Type	Description	Default
request	Request	The incoming HTTP request.	required
call_next	Callable[[Request], Awaitable[Response]]	The next interceptor or endpoint to call.	required

Returns:

Name	Type	Description
Response	Response	The HTTP response from the endpoint.

Raises:

Type	Description
Exception	If an exception occurs during request processing, it is captured and re-raised.

Source code in archipy/helpers/interceptors/fastapi/metric/interceptor.py

async def dispatch(self, request: Request, call_next: Callable[[Request], Awaitable[Response]]) -> Response:
    """Intercept HTTP requests to measure response time and track active requests.

    Args:
        request (Request): The incoming HTTP request.
        call_next (Callable[[Request], Awaitable[Response]]): The next interceptor or endpoint to call.

    Returns:
        Response: The HTTP response from the endpoint.

    Raises:
        Exception: If an exception occurs during request processing, it is captured and re-raised.
    """
    if not BaseConfig.global_config().PROMETHEUS.IS_ENABLED:
        return await call_next(request)

    path_template = self._get_path_template(request)
    method = request.method

    self.ACTIVE_REQUESTS.labels(method=method, path_template=path_template).inc()

    start_time = time.time()
    status_code = 500

    try:
        response = await call_next(request)
        status_code = response.status_code
    except Exception as exception:
        BaseUtils.capture_exception(exception)
        raise
    else:
        return response
    finally:
        duration = time.time() - start_time
        self.RESPONSE_TIME_SECONDS.labels(
            method=method,
            status_code=status_code,
            path_template=path_template,
        ).observe(duration)
        self.ACTIVE_REQUESTS.labels(method=method, path_template=path_template).dec()

options: show_root_toc_entry: false heading_level: 3

rate_limit

FastAPI interceptor that enforces configurable rate limits on HTTP endpoints using Redis as a backend.

archipy.helpers.interceptors.fastapi.rate_limit.fastapi_rest_rate_limit_handler.FastAPIRestRateLimitHandler

A rate-limiting handler for FastAPI REST endpoints using Redis for tracking.

This class provides rate-limiting functionality by tracking the number of requests made to a specific endpoint within a defined time window. If the request limit is exceeded, it raises an HTTP 429 Too Many Requests error.

Parameters:

Name	Type	Description	Default
calls_count	StrictInt	The maximum number of allowed requests within the time window.	1
milliseconds	StrictInt	The time window in milliseconds.	0
seconds	StrictInt	The time window in seconds.	0
minutes	StrictInt	The time window in minutes.	0
hours	StrictInt	The time window in hours.	0
days	StrictInt	The time window in days.	0
query_params	set(StrictStr)	request query parameters for rate-limiting based on query params.	None

Source code in archipy/helpers/interceptors/fastapi/rate_limit/fastapi_rest_rate_limit_handler.py

class FastAPIRestRateLimitHandler:
    """A rate-limiting handler for FastAPI REST endpoints using Redis for tracking.

    This class provides rate-limiting functionality by tracking the number of requests
    made to a specific endpoint within a defined time window. If the request limit is
    exceeded, it raises an HTTP 429 Too Many Requests error.

    Args:
        calls_count (StrictInt): The maximum number of allowed requests within the time window.
        milliseconds (StrictInt): The time window in milliseconds.
        seconds (StrictInt): The time window in seconds.
        minutes (StrictInt): The time window in minutes.
        hours (StrictInt): The time window in hours.
        days (StrictInt): The time window in days.
        query_params (set(StrictStr)): request query parameters for rate-limiting based on query params.
    """

    def __init__(
        self,
        calls_count: StrictInt = 1,
        milliseconds: StrictInt = 0,
        seconds: StrictInt = 0,
        minutes: StrictInt = 0,
        hours: StrictInt = 0,
        days: StrictInt = 0,
        query_params: set[StrictStr] | None = None,
        *,
        rate_limit_config: FastAPIRateLimitConfig | None = None,
        trusted_proxy_ips: list[str] | None = None,
        allow_private_ips: bool | None = None,
        identifier_fn: Callable[[Request], str] | None = None,
        key_prefix: str | None = None,
        fail_closed: bool | None = None,
        skip_methods: frozenset[str] | None = None,
        hash_query_param_values: bool | None = None,
        require_identifier_for_query_params: bool | None = None,
        reject_unknown_client: bool | None = None,
        rate_limit_headers: bool | None = None,
        identity_from_access_token: bool | None = None,
        additional_windows: list[RateLimitWindowDTO] | None = None,
    ) -> None:
        """Initialize the rate limit handler with specified time window and request limits.

        The time window is calculated by combining all time unit parameters into milliseconds.
        At least one time unit parameter should be greater than 0 to create a valid window.

        Args:
            calls_count (StrictInt, optional): Maximum number of allowed requests within the time window.
                Defaults to 1.
            milliseconds (StrictInt, optional): Number of milliseconds in the time window.
                Defaults to 0.
            seconds (StrictInt, optional): Number of seconds in the time window.
                Defaults to 0.
            minutes (StrictInt, optional): Number of minutes in the time window.
                Defaults to 0.
            hours (StrictInt, optional): Number of hours in the time window.
                Defaults to 0.
            days (StrictInt, optional): Number of days in the time window.
                Defaults to 0.
            query_params (set[StrictStr] | None, optional): Set of query parameter names to include
                in rate limit key generation. If None, no query parameters will be used.
                Defaults to None.
            rate_limit_config (FastAPIRateLimitConfig | None, optional): Rate-limit settings.
                When None, ``FASTAPI_RATE_LIMIT`` from ``BaseConfig.global_config()`` is used.
            trusted_proxy_ips (list[str] | None, optional): Override for
                ``FASTAPI_RATE_LIMIT.TRUSTED_PROXY_IPS`` / ``FASTAPI.FORWARDED_ALLOW_IPS``.
            allow_private_ips (bool | None, optional): Override for
                ``FASTAPI_RATE_LIMIT.ALLOW_PRIVATE_IPS``.
            identifier_fn (Callable[[Request], str] | None, optional): Custom callable that returns
                the rate-limit identity (for example authenticated user ID). When provided, it
                replaces IP-based identification.
            key_prefix (str | None, optional): Override for ``FASTAPI_RATE_LIMIT.KEY_PREFIX``.
            fail_closed (bool | None, optional): Override for ``FASTAPI_RATE_LIMIT.FAIL_CLOSED``.
            skip_methods (frozenset[str] | None, optional): Override for
                ``FASTAPI_RATE_LIMIT.SKIP_METHODS``.
            hash_query_param_values (bool | None, optional): Override for
                ``FASTAPI_RATE_LIMIT.HASH_QUERY_PARAM_VALUES``.
            require_identifier_for_query_params (bool | None, optional): Override for
                ``FASTAPI_RATE_LIMIT.REQUIRE_IDENTIFIER_FOR_QUERY_PARAMS``.
            reject_unknown_client (bool | None, optional): Override for
                ``FASTAPI_RATE_LIMIT.REJECT_UNKNOWN_CLIENT``.
            rate_limit_headers (bool | None, optional): Override for
                ``FASTAPI_RATE_LIMIT.RATE_LIMIT_HEADERS``.
            identity_from_access_token (bool | None, optional): Override for
                ``FASTAPI_RATE_LIMIT.IDENTITY_FROM_ACCESS_TOKEN``. Defaults to True. When enabled,
                verifies Bearer access tokens via ``JWTUtils`` and buckets by ``sub``, falling back
                to client IP.
            additional_windows (list[RateLimitWindowDTO] | None, optional): Extra rate-limit tiers
                enforced in addition to the primary window built from ``calls_count`` and time units.
                All windows must pass; the most constrained remaining quota is reported in headers.

        Raises:
            InvalidArgumentError: If ``calls_count``, the computed window, or ``query_params`` config is invalid.

        Example:
            >>> # Allow 100 requests per minute
            >>> handler = FastAPIRestRateLimitHandler(calls_count=100, minutes=1)
            >>>
            >>> # Allow 1000 requests per day with specific query params
            >>> handler = FastAPIRestRateLimitHandler(calls_count=1000, days=1, query_params={"user_id", "action"})
        """
        primary_window = RateLimitUtils.compute_rate_limit_window(
            calls_count=calls_count,
            milliseconds=milliseconds,
            seconds=seconds,
            minutes=minutes,
            hours=hours,
            days=days,
        )
        self._windows: tuple[RateLimitWindowDTO, ...] = (primary_window, *(additional_windows or ()))
        self.calls_count = primary_window.calls_count
        self.milliseconds = primary_window.window_ms

        self.query_params = query_params or set()
        resolved_rate_limit_config = rate_limit_config or BaseConfig.global_config().FASTAPI_RATE_LIMIT
        resolved_require_identifier = (
            require_identifier_for_query_params
            if require_identifier_for_query_params is not None
            else resolved_rate_limit_config.REQUIRE_IDENTIFIER_FOR_QUERY_PARAMS
        )
        resolved_identity_from_access_token = (
            identity_from_access_token
            if identity_from_access_token is not None
            else resolved_rate_limit_config.IDENTITY_FROM_ACCESS_TOKEN
        )
        has_server_resolved_identity = identifier_fn is not None or resolved_identity_from_access_token
        if resolved_require_identifier and self.query_params and not has_server_resolved_identity:
            raise InvalidArgumentError(
                additional_data={
                    "detail": (
                        "query_params requires identifier_fn or identity_from_access_token to avoid "
                        "client-controlled quota targeting; set require_identifier_for_query_params=False "
                        "to override"
                    ),
                },
            )

        app_config = BaseConfig.global_config()
        resolved_trusted_proxy_ips = trusted_proxy_ips
        if resolved_trusted_proxy_ips is None:
            resolved_trusted_proxy_ips = resolved_rate_limit_config.TRUSTED_PROXY_IPS
        if resolved_trusted_proxy_ips is None:
            resolved_trusted_proxy_ips = app_config.FASTAPI.FORWARDED_ALLOW_IPS or []

        self._always_trust_proxies, self._trusted_proxy_exact, self._trusted_proxy_networks = (
            self._parse_trusted_proxy_ips(resolved_trusted_proxy_ips)
        )
        self._allow_private_ips = (
            allow_private_ips if allow_private_ips is not None else resolved_rate_limit_config.ALLOW_PRIVATE_IPS
        )
        self._identifier_fn = identifier_fn
        self._identity_from_access_token = resolved_identity_from_access_token
        resolved_key_prefix = key_prefix if key_prefix is not None else resolved_rate_limit_config.KEY_PREFIX
        self._key_prefix = resolved_key_prefix or f"{app_config.FASTAPI.PROJECT_NAME}:RateLimit"
        self._fail_closed = fail_closed if fail_closed is not None else resolved_rate_limit_config.FAIL_CLOSED
        resolved_skip_methods = (
            skip_methods if skip_methods is not None else frozenset(resolved_rate_limit_config.SKIP_METHODS)
        )
        self._skip_methods = resolved_skip_methods
        self._hash_query_param_values = (
            hash_query_param_values
            if hash_query_param_values is not None
            else resolved_rate_limit_config.HASH_QUERY_PARAM_VALUES
        )
        self._reject_unknown_client = (
            reject_unknown_client
            if reject_unknown_client is not None
            else resolved_rate_limit_config.REJECT_UNKNOWN_CLIENT
        )
        self._rate_limit_headers = (
            rate_limit_headers if rate_limit_headers is not None else resolved_rate_limit_config.RATE_LIMIT_HEADERS
        )
        self._redis_client: AsyncRedisAdapter = self._create_redis_client()

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

        return AsyncRedisAdapter()

    @staticmethod
    def _parse_trusted_proxy_ips(
        trusted_proxy_ips: list[str],
    ) -> tuple[bool, frozenset[str], tuple[IPv4Network | IPv6Network, ...]]:
        """Parse trusted proxy IP entries into always-trust flag, exact hosts, and networks."""
        always_trust = False
        exact_hosts: set[str] = set()
        networks = []
        for entry in trusted_proxy_ips:
            stripped = entry.strip()
            if not stripped:
                continue
            if stripped == "*":
                always_trust = True
                continue
            try:
                networks.append(ip_network(stripped, strict=False))
            except ValueError:
                exact_hosts.add(stripped)
        return always_trust, frozenset(exact_hosts), tuple(networks)

    def _is_trusted_proxy_value(self, host: str) -> bool:
        """Return whether a host or IP belongs to the configured trusted proxy set."""
        if self._always_trust_proxies:
            return bool(host)
        if not host or (not self._trusted_proxy_exact and not self._trusted_proxy_networks):
            return False
        if host in self._trusted_proxy_exact:
            return True
        try:
            ip = ip_address(host.strip())
        except ValueError:
            return False
        return any(ip in network for network in self._trusted_proxy_networks)

    def _peer_is_trusted(self, peer_host: str | None) -> bool:
        """Return whether the immediate peer is a configured trusted proxy."""
        if not peer_host:
            return False
        if self._always_trust_proxies:
            return True
        return self._is_trusted_proxy_value(peer_host)

    @staticmethod
    def _is_non_routable(ip: IPv4Address | IPv6Address) -> bool:
        """Return whether an IP should be rejected when private addresses are disallowed."""
        return ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_multicast

    def _normalize_ip(self, ip_str: str) -> str | None:
        """Validate and canonicalize an IP address string."""
        host = self._parse_host_port(ip_str)
        try:
            ip = ip_address(host)
        except ValueError:
            return None
        if not self._allow_private_ips and self._is_non_routable(ip):
            return None
        return str(ip)

    @staticmethod
    def _parse_host_port(value: str) -> str:
        """Return the host portion from a forwarded address token that may include a port."""
        stripped = value.strip()
        if stripped.startswith("[") and "]" in stripped:
            return stripped[1 : stripped.index("]")]
        if stripped.count(":") == 1 and "." in stripped:
            return stripped.split(":", maxsplit=1)[0]
        return stripped

    @staticmethod
    def _hash_segment(value: str) -> str:
        """Return a fixed-length hash for Redis key segments."""
        return hashlib.sha256(value.encode()).hexdigest()[:16]

    @staticmethod
    def _normalize_path(path: str) -> str:
        """Normalize request paths for stable rate-limit keys."""
        if path != "/" and path.endswith("/"):
            return path.rstrip("/")
        return path

    @staticmethod
    def _get_combined_x_forwarded_for(request: Request) -> str | None:
        """Combine all X-Forwarded-For header fields per RFC 9110."""
        values = request.headers.getlist("x-forwarded-for")
        if not values:
            return None
        return ", ".join(value.strip() for value in values if value.strip())

    def _resolve_trusted_chain(self, entries: list[str]) -> str | None:
        """Resolve the client IP from an ordered forwarded chain using recursive trust semantics."""
        if not entries:
            return None

        hosts = [self._parse_host_port(entry) for entry in entries]
        if self._always_trust_proxies:
            return self._normalize_ip(hosts[0])

        for host in reversed(hosts):
            candidate = self._normalize_ip(host) or host
            if not self._is_trusted_proxy_value(candidate):
                return self._normalize_ip(host)

        return self._normalize_ip(hosts[0])

    async def _check_window(self, key: str, window: RateLimitWindowDTO) -> tuple[int, int]:
        """Increment one rate-limit counter and return limit state.

        Returns:
            tuple[int, int]: ``(pexpire_if_limited, remaining_calls)``. ``pexpire`` is 0 when allowed.
        """
        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 _check_windows(self, base_key: str) -> tuple[RateLimitWindowDTO | None, int, int, RateLimitWindowDTO]:
        """Evaluate all configured windows for a request.

        Returns:
            tuple containing the breaching window (or None), PTTL if limited, remaining calls for
            the header window, and the header window itself.
        """
        header_window = self._windows[0]
        header_remaining = 0
        min_remaining_ratio = 1.0

        for window in self._windows:
            key = f"{base_key}:{window.key_suffix}"
            pexpire, remaining = await self._check_window(key, window)
            if pexpire != 0:
                return window, pexpire, 0, window

            remaining_ratio = remaining / window.calls_count
            if remaining_ratio <= min_remaining_ratio:
                min_remaining_ratio = remaining_ratio
                header_window = window
                header_remaining = remaining

        return None, 0, header_remaining, header_window

    def _rate_limit_response_headers(
        self,
        remaining: int,
        *,
        limited: bool,
        calls_count: int,
        pexpire: int = 0,
        window_ms: int | None = None,
    ) -> dict[str, str]:
        """Build standard rate-limit response headers."""
        headers = {
            "X-RateLimit-Limit": str(calls_count),
            "X-RateLimit-Remaining": str(remaining),
        }
        if limited:
            headers["Retry-After"] = str(self._retry_after_seconds(pexpire, window_ms or self.milliseconds))
        return headers

    async def __call__(self, request: Request, response: Response) -> None:
        """Handles the rate-limiting logic for incoming requests.

        Args:
            request (Request): The incoming FastAPI request.
            response (Response): The outgoing response for rate-limit headers.

        Raises:
            HTTPException: If the rate limit is exceeded, an HTTP 429 Too Many Requests error is raised.
            HTTPException: If Redis is unavailable and ``fail_closed`` is enabled.
        """
        from redis.exceptions import RedisError  # noqa: PLC0415

        if request.method in self._skip_methods:
            return

        rate_key = self._get_identifier(request)
        base_key = f"{self._key_prefix}:{rate_key}:{request.method}"
        try:
            breaching_window, pexpire, remaining, header_window = await self._check_windows(base_key)
        except (ConnectionError, OSError, TimeoutError, RedisError) as exc:
            if self._fail_closed:
                raise HTTPException(
                    status_code=HTTP_503_SERVICE_UNAVAILABLE,
                    detail="Rate limiter unavailable",
                ) from exc
            return

        if breaching_window is not None:
            self._create_callback(pexpire, breaching_window)

        if self._rate_limit_headers:
            for header_name, header_value in self._rate_limit_response_headers(
                remaining,
                limited=False,
                calls_count=header_window.calls_count,
            ).items():
                response.headers[header_name] = header_value

    def _retry_after_seconds(self, pexpire: int, window_ms: int) -> int:
        """Convert a PTTL value to a safe Retry-After header in seconds."""
        if pexpire > 0:
            return max(1, ceil(pexpire / 1000))
        return max(1, ceil(window_ms / 1000))

    def _create_callback(self, pexpire: int, window: RateLimitWindowDTO) -> None:
        """Raises an HTTP 429 Too Many Requests error with the appropriate headers.

        Args:
            pexpire (int): The remaining time-to-live (TTL) in milliseconds before the rate limit resets.
            window (RateLimitWindowDTO): The breached rate-limit window.

        Raises:
            HTTPException: An HTTP 429 Too Many Requests error with rate-limit headers.
        """
        headers = (
            self._rate_limit_response_headers(
                0,
                limited=True,
                calls_count=window.calls_count,
                pexpire=pexpire,
                window_ms=window.window_ms,
            )
            if self._rate_limit_headers
            else {"Retry-After": str(self._retry_after_seconds(pexpire, window.window_ms))}
        )
        raise HTTPException(
            status_code=HTTP_429_TOO_MANY_REQUESTS,
            detail="Too Many Requests",
            headers=headers,
        )

    def _resolve_base_identifier(self, request: Request) -> str:
        """Resolve the base rate-limit identity from the request.

        Args:
            request: The incoming FastAPI request.

        Returns:
            str: Server-resolved client or user identity.

        Raises:
            HTTPException: If identity resolution fails and ``fail_closed`` is enabled.
        """
        if self._identifier_fn is not None:
            try:
                return self._identifier_fn(request)
            except Exception as exc:
                if self._fail_closed:
                    raise HTTPException(
                        status_code=HTTP_503_SERVICE_UNAVAILABLE,
                        detail="Rate limiter identity resolution failed",
                    ) from exc
                return "unknown"

        if self._identity_from_access_token:
            if user_sub := resolve_jwt_access_token_sub(request):
                return user_sub
            return self._extract_client_ip(request)

        return self._extract_client_ip(request)

    def _get_identifier(self, request: Request) -> str:
        """Extract a unique identifier from the request, typically an IP address and endpoint.

        Args:
            request (Request): The FastAPI request object containing headers and client info.

        Returns:
            str: A Redis key segment built from the client identity and request path.

        Raises:
            HTTPException: If identity resolution fails and ``fail_closed`` is enabled.
        """
        base_identifier = self._resolve_base_identifier(request)

        path = self._normalize_path(request.url.path)
        path_key = f"{base_identifier}:{path}"

        if not self.query_params:
            return path_key

        return self._append_query_params(path_key, request.query_params)

    def _extract_client_ip(self, request: Request) -> str:
        """Extract and validate the client IP from the request.

        Forwarded headers are honored only when the immediate peer is a configured trusted proxy.

        Args:
            request (Request): The FastAPI request object.

        Returns:
            str: Canonical client IP or peer host.

        Raises:
            HTTPException: If the client cannot be identified and ``reject_unknown_client`` is enabled.
        """
        peer = request.client.host if request.client is not None else None

        if peer and self._peer_is_trusted(peer):
            if client_ip := self._client_ip_from_trusted_headers(request):
                return client_ip
            if normalized_peer := self._normalize_ip(peer):
                return normalized_peer
            return peer

        if peer:
            if normalized_peer := self._normalize_ip(peer):
                return normalized_peer
            return peer

        if self._reject_unknown_client:
            raise HTTPException(
                status_code=HTTP_503_SERVICE_UNAVAILABLE,
                detail="Client identity unknown",
            )
        return "unknown"

    def _client_ip_from_trusted_headers(self, request: Request) -> str | None:
        """Resolve the client IP from proxy headers when the peer is trusted."""
        for header_name in _SINGLE_VALUE_PROXY_HEADER_NAMES:
            if client_ip := self._validate_ip_from_header(request.headers.get(header_name)):
                return client_ip

        if forwarded := request.headers.get("Forwarded"):
            if client_ip := self._parse_forwarded_header(forwarded):
                return client_ip

        if forwarded_for := self._get_combined_x_forwarded_for(request):
            if client_ip := self._client_ip_from_forwarded_for(forwarded_for):
                return client_ip

        if client_ip := self._validate_ip_from_header(request.headers.get("X-Real-IP")):
            return client_ip

        return None

    def _validate_ip_from_header(self, header_value: str | None) -> str | None:
        """Validate the first IP in a single-value proxy header.

        Args:
            header_value (str | None): IP address from a proxy header.

        Returns:
            str | None: Canonical IP address or None.
        """
        if not header_value:
            return None

        first_value = header_value.split(",")[0].strip()
        return self._normalize_ip(first_value)

    def _client_ip_from_forwarded_for(self, forwarded_for: str) -> str | None:
        """Validate the client IP from an X-Forwarded-For header chain.

        Walks the chain right-to-left (nginx ``real_ip_recursive`` semantics): trusted proxy
        entries are skipped until the first non-trusted address is found.

        Args:
            forwarded_for (str): X-Forwarded-For header value.

        Returns:
            str | None: Canonical client IP or None.
        """
        parts = [part.strip() for part in forwarded_for.split(",") if part.strip()]
        return self._resolve_trusted_chain(parts)

    def _parse_forwarded_header(self, forwarded_value: str) -> str | None:
        """Parse the client IP from an RFC 7239 Forwarded header using recursive trust semantics.

        Args:
            forwarded_value (str): Forwarded header value.

        Returns:
            str | None: Canonical client IP or None.
        """
        forwarded_hosts: list[str] = []
        for forwarded_part in forwarded_value.split(","):
            for directive in forwarded_part.split(";"):
                stripped = directive.strip()
                if not stripped.lower().startswith("for="):
                    continue

                for_value = stripped[4:].strip().strip('"')
                if for_value.lower() == "unknown":
                    continue
                forwarded_hosts.append(for_value)

        return self._resolve_trusted_chain(forwarded_hosts)

    def _append_query_params(self, base_key: str, query_params: QueryParams) -> str:
        """Append sorted query parameters to the Redis key.

        Args:
            base_key (str): Base Redis key without query parameters.
            query_params (QueryParams): Request query parameters.

        Returns:
            str: Redis key with appended query parameters.
        """
        filtered_params = {
            key: value for key, value in query_params.items() if key in self.query_params and value is not None
        }

        if not filtered_params:
            return base_key

        sorted_params = sorted(filtered_params.items())
        if self._hash_query_param_values:
            query_string = "&".join(f"{key}={self._hash_segment(value)}" for key, value in sorted_params)
        else:
            query_string = "&".join(f"{key}={value}" for key, value in sorted_params)
        return f"{base_key}?{query_string}"

archipy.helpers.interceptors.fastapi.rate_limit.fastapi_rest_rate_limit_handler.FastAPIRestRateLimitHandler.calls_count instance-attribute

calls_count = primary_window.calls_count

archipy.helpers.interceptors.fastapi.rate_limit.fastapi_rest_rate_limit_handler.FastAPIRestRateLimitHandler.milliseconds instance-attribute

milliseconds = primary_window.window_ms

archipy.helpers.interceptors.fastapi.rate_limit.fastapi_rest_rate_limit_handler.FastAPIRestRateLimitHandler.query_params instance-attribute

query_params = query_params or set()

options: show_root_toc_entry: false heading_level: 3

gRPC

base

Abstract base classes for gRPC client and server interceptors.

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.
        """
        result = self.intercept(_swap_args(continuation), request, client_call_details)
        return result

    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.
        """
        result = self.intercept(_swap_args(continuation), request, client_call_details)
        return result

    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.
        """
        result = self.intercept(_swap_args(continuation), request_iterator, client_call_details)
        return result

    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.
        """
        result = self.intercept(_swap_args(continuation), request_iterator, client_call_details)
        return result

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.
    """
    result = self.intercept(_swap_args(continuation), request, client_call_details)
    return result

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.
    """
    result = self.intercept(_swap_args(continuation), request, client_call_details)
    return result

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.
    """
    result = self.intercept(_swap_args(continuation), request_iterator, client_call_details)
    return result

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.
    """
    result = self.intercept(_swap_args(continuation), request_iterator, client_call_details)
    return result

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.
        """
        result = await self.intercept(_swap_args(continuation), request, client_call_details)
        return result

    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.
        """
        result = await self.intercept(_swap_args(continuation), request, client_call_details)
        return result

    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.
        """
        result = await self.intercept(_swap_args(continuation), request_iterator, client_call_details)
        return result

    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.
        """
        result = await self.intercept(_swap_args(continuation), request_iterator, client_call_details)
        return result

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.
    """
    result = await self.intercept(_swap_args(continuation), request, client_call_details)
    return result

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.
    """
    result = await self.intercept(_swap_args(continuation), request, client_call_details)
    return result

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.
    """
    result = await self.intercept(_swap_args(continuation), request_iterator, client_call_details)
    return result

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.
    """
    result = await self.intercept(_swap_args(continuation), request_iterator, client_call_details)
    return result

options: show_root_toc_entry: false heading_level: 3

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.

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)

    @staticmethod
    def _format_validation_errors(validation_error: ValidationError) -> list[dict[str, str]]:
        """Format Pydantic validation errors into a structured format.

        Args:
            validation_error: The validation error to format.

        Returns:
            A list of formatted validation error details.

        Note:
            This method is deprecated. Use BaseUtils.format_validation_errors instead.
        """
        return BaseUtils.format_validation_errors(validation_error, include_type=True)

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)

    @staticmethod
    def _format_validation_errors(validation_error: ValidationError) -> list[dict[str, str]]:
        """Format Pydantic validation errors into a structured format.

        Args:
            validation_error: The validation error to format.

        Returns:
            A list of formatted validation error details.

        Note:
            This method is deprecated. Use BaseUtils.format_validation_errors instead.
        """
        return BaseUtils.format_validation_errors(validation_error, include_type=True)

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

metric

gRPC server interceptor for collecting Prometheus metrics on RPC calls.

archipy.helpers.interceptors.grpc.metric.server_interceptor.GrpcServerMetricInterceptor

Bases: BaseGrpcServerInterceptor

A gRPC server interceptor for collecting and reporting metrics using Prometheus.

This interceptor measures the response time of gRPC methods and records it in a Prometheus histogram. It also tracks the number of active requests using a Prometheus gauge. It also captures errors and logs them for monitoring purposes.

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

class GrpcServerMetricInterceptor(BaseGrpcServerInterceptor):
    """A gRPC server interceptor for collecting and reporting metrics using Prometheus.

    This interceptor measures the response time of gRPC methods and records it in a Prometheus histogram.
    It also tracks the number of active requests using a Prometheus gauge.
    It also captures errors and logs them for monitoring purposes.
    """

    from prometheus_client import Gauge, Histogram

    "Buckets for measuring response times between 0 and 1 second."
    ZERO_TO_ONE_SECONDS_BUCKETS: ClassVar[list[float]] = [i / 1000 for i in range(0, 1000, 5)]

    "Buckets for measuring response times between 1 and 5 seconds."
    ONE_TO_FIVE_SECONDS_BUCKETS: ClassVar[list[float]] = [i / 100 for i in range(100, 500, 20)]

    "Buckets for measuring response times between 5 and 30 seconds."
    FIVE_TO_THIRTY_SECONDS_BUCKETS: ClassVar[list[float]] = [i / 100 for i in range(500, 3000, 50)]

    "Combined buckets for measuring response times from 0 to 30 seconds and beyond."
    TOTAL_BUCKETS = (
        ZERO_TO_ONE_SECONDS_BUCKETS + ONE_TO_FIVE_SECONDS_BUCKETS + FIVE_TO_THIRTY_SECONDS_BUCKETS + [float("inf")]
    )

    "Prometheus histogram for tracking response times of gRPC methods."
    RESPONSE_TIME_SECONDS = Histogram(
        "grpc_response_time_seconds",
        "Time spent processing gRPC request",
        labelnames=("package", "service", "method", "status_code"),
        buckets=TOTAL_BUCKETS,
    )

    "Prometheus gauge for tracking active gRPC requests."
    ACTIVE_REQUESTS = Gauge(
        "grpc_active_requests",
        "Number of active gRPC requests",
        labelnames=("package", "service", "method"),
    )

    def intercept(
        self,
        method: Callable,
        request: object,
        context: grpc.ServicerContext,
        method_name_model: MethodName,
    ) -> object:
        """Intercepts a gRPC server call to measure response time and track active requests.

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

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

        Raises:
            Exception: If an exception occurs during the method execution, it is captured and logged.
        """
        if not BaseConfig.global_config().PROMETHEUS.IS_ENABLED:
            return method(request, context)

        package = method_name_model.package
        service = method_name_model.service
        method_name = method_name_model.method

        self.ACTIVE_REQUESTS.labels(package=package, service=service, method=method_name).inc()

        start_time = time.time()
        status_code = "OK"

        try:
            result = method(request, context)

            if hasattr(context, "code") and callable(context.code):
                code_method = context.code
                code_obj = code_method()  # ty: ignore[call-top-callable]
                if code_obj is not None:
                    code_name = getattr(code_obj, "name", None)
                    if code_name is not None:
                        status_code = code_name
        except Exception as exception:
            BaseUtils.capture_exception(exception)
            raise
        else:
            return result
        finally:
            duration = time.time() - start_time
            self.RESPONSE_TIME_SECONDS.labels(
                package=package,
                service=service,
                method=method_name,
                status_code=status_code,
            ).observe(duration)
            self.ACTIVE_REQUESTS.labels(package=package, service=service, method=method_name).dec()

archipy.helpers.interceptors.grpc.metric.server_interceptor.GrpcServerMetricInterceptor.ZERO_TO_ONE_SECONDS_BUCKETS class-attribute

ZERO_TO_ONE_SECONDS_BUCKETS: list[float] = [
    (i / 1000) for i in (range(0, 1000, 5))
]

Buckets for measuring response times between 1 and 5 seconds.

archipy.helpers.interceptors.grpc.metric.server_interceptor.GrpcServerMetricInterceptor.ONE_TO_FIVE_SECONDS_BUCKETS class-attribute

ONE_TO_FIVE_SECONDS_BUCKETS: list[float] = [
    (i / 100) for i in (range(100, 500, 20))
]

Buckets for measuring response times between 5 and 30 seconds.

archipy.helpers.interceptors.grpc.metric.server_interceptor.GrpcServerMetricInterceptor.FIVE_TO_THIRTY_SECONDS_BUCKETS class-attribute

FIVE_TO_THIRTY_SECONDS_BUCKETS: list[float] = [
    (i / 100) for i in (range(500, 3000, 50))
]

Combined buckets for measuring response times from 0 to 30 seconds and beyond.

archipy.helpers.interceptors.grpc.metric.server_interceptor.GrpcServerMetricInterceptor.TOTAL_BUCKETS class-attribute instance-attribute

TOTAL_BUCKETS = (
    ZERO_TO_ONE_SECONDS_BUCKETS
    + ONE_TO_FIVE_SECONDS_BUCKETS
    + FIVE_TO_THIRTY_SECONDS_BUCKETS
    + [float("inf")]
)

Prometheus histogram for tracking response times of gRPC methods.

archipy.helpers.interceptors.grpc.metric.server_interceptor.GrpcServerMetricInterceptor.RESPONSE_TIME_SECONDS class-attribute instance-attribute

RESPONSE_TIME_SECONDS = Histogram(
    "grpc_response_time_seconds",
    "Time spent processing gRPC request",
    labelnames=(
        "package",
        "service",
        "method",
        "status_code",
    ),
    buckets=TOTAL_BUCKETS,
)

Prometheus gauge for tracking active gRPC requests.

archipy.helpers.interceptors.grpc.metric.server_interceptor.GrpcServerMetricInterceptor.ACTIVE_REQUESTS class-attribute instance-attribute

ACTIVE_REQUESTS = Gauge(
    "grpc_active_requests",
    "Number of active gRPC requests",
    labelnames=("package", "service", "method"),
)

archipy.helpers.interceptors.grpc.metric.server_interceptor.GrpcServerMetricInterceptor.intercept

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

Intercepts a gRPC server call to measure response time and track active requests.

Parameters:

Name	Type	Description	Default
method	Callable	The gRPC method being intercepted.	required
request	object	The request object passed to the method.	required
context	ServicerContext	The context of the 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.

Raises:

Type	Description
Exception	If an exception occurs during the method execution, it is captured and logged.

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

def intercept(
    self,
    method: Callable,
    request: object,
    context: grpc.ServicerContext,
    method_name_model: MethodName,
) -> object:
    """Intercepts a gRPC server call to measure response time and track active requests.

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

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

    Raises:
        Exception: If an exception occurs during the method execution, it is captured and logged.
    """
    if not BaseConfig.global_config().PROMETHEUS.IS_ENABLED:
        return method(request, context)

    package = method_name_model.package
    service = method_name_model.service
    method_name = method_name_model.method

    self.ACTIVE_REQUESTS.labels(package=package, service=service, method=method_name).inc()

    start_time = time.time()
    status_code = "OK"

    try:
        result = method(request, context)

        if hasattr(context, "code") and callable(context.code):
            code_method = context.code
            code_obj = code_method()  # ty: ignore[call-top-callable]
            if code_obj is not None:
                code_name = getattr(code_obj, "name", None)
                if code_name is not None:
                    status_code = code_name
    except Exception as exception:
        BaseUtils.capture_exception(exception)
        raise
    else:
        return result
    finally:
        duration = time.time() - start_time
        self.RESPONSE_TIME_SECONDS.labels(
            package=package,
            service=service,
            method=method_name,
            status_code=status_code,
        ).observe(duration)
        self.ACTIVE_REQUESTS.labels(package=package, service=service, method=method_name).dec()

archipy.helpers.interceptors.grpc.metric.server_interceptor.GrpcServerMetricInterceptor.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.metric.server_interceptor.AsyncGrpcServerMetricInterceptor

Bases: BaseAsyncGrpcServerInterceptor

An async gRPC server interceptor for collecting and reporting metrics using Prometheus.

This interceptor measures the response time of async gRPC methods and records it in a Prometheus histogram. It also tracks the number of active requests using a Prometheus gauge. It also captures errors and logs them for monitoring purposes.

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

class AsyncGrpcServerMetricInterceptor(BaseAsyncGrpcServerInterceptor):
    """An async gRPC server interceptor for collecting and reporting metrics using Prometheus.

    This interceptor measures the response time of async gRPC methods and records it in a Prometheus histogram.
    It also tracks the number of active requests using a Prometheus gauge.
    It also captures errors and logs them for monitoring purposes.
    """

    from prometheus_client import Gauge, Histogram

    "Buckets for measuring response times between 0 and 1 second."
    ZERO_TO_ONE_SECONDS_BUCKETS: ClassVar[list[float]] = [i / 1000 for i in range(0, 1000, 5)]

    "Buckets for measuring response times between 1 and 5 seconds."
    ONE_TO_FIVE_SECONDS_BUCKETS: ClassVar[list[float]] = [i / 100 for i in range(100, 500, 20)]

    "Buckets for measuring response times between 5 and 30 seconds."
    FIVE_TO_THIRTY_SECONDS_BUCKETS: ClassVar[list[float]] = [i / 100 for i in range(500, 3000, 50)]

    "Combined buckets for measuring response times from 0 to 30 seconds and beyond."
    TOTAL_BUCKETS = (
        ZERO_TO_ONE_SECONDS_BUCKETS + ONE_TO_FIVE_SECONDS_BUCKETS + FIVE_TO_THIRTY_SECONDS_BUCKETS + [float("inf")]
    )

    "Prometheus histogram for tracking response times of async gRPC methods."
    RESPONSE_TIME_SECONDS = Histogram(
        "grpc_async_response_time_seconds",
        "Time spent processing async gRPC request",
        labelnames=("package", "service", "method", "status_code"),
        buckets=TOTAL_BUCKETS,
    )

    "Prometheus gauge for tracking active async gRPC requests."
    ACTIVE_REQUESTS = Gauge(
        "grpc_async_active_requests",
        "Number of active async gRPC requests",
        labelnames=("package", "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 to measure response time and track active requests.

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

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

        Raises:
            Exception: If an exception occurs during the method execution, it is captured and logged.
        """
        if not BaseConfig.global_config().PROMETHEUS.IS_ENABLED:
            return await method(request, context)

        package = method_name_model.package
        service = method_name_model.service
        method_name = method_name_model.method

        self.ACTIVE_REQUESTS.labels(package=package, service=service, method=method_name).inc()

        start_time = asyncio.get_event_loop().time()
        status_code = "OK"

        try:
            try:
                result = await method(request, context)

                if hasattr(context, "code") and context.code():
                    status_code = context.code().name
            except Exception as exception:
                if isinstance(exception, grpc.aio.AioRpcError):
                    code_obj = exception.code()
                    if code_obj is not None:
                        code_name = getattr(code_obj, "name", None)
                        if code_name is not None:
                            status_code = code_name
                elif hasattr(exception, "code") and callable(exception.code):
                    code_method = exception.code
                    code_obj = code_method()  # ty: ignore[call-top-callable]
                    if code_obj is not None:
                        code_name = getattr(code_obj, "name", None)
                        if code_name is not None:
                            status_code = code_name
                else:
                    status_code = "INTERNAL"
                raise
            else:
                return result
            finally:
                duration = asyncio.get_event_loop().time() - start_time
                self.RESPONSE_TIME_SECONDS.labels(
                    package=package,
                    service=service,
                    method=method_name,
                    status_code=status_code,
                ).observe(duration)
                self.ACTIVE_REQUESTS.labels(package=package, service=service, method=method_name).dec()

        except Exception as exception:
            BaseUtils.capture_exception(exception)
            raise

archipy.helpers.interceptors.grpc.metric.server_interceptor.AsyncGrpcServerMetricInterceptor.ZERO_TO_ONE_SECONDS_BUCKETS class-attribute

ZERO_TO_ONE_SECONDS_BUCKETS: list[float] = [
    (i / 1000) for i in (range(0, 1000, 5))
]

Buckets for measuring response times between 1 and 5 seconds.

archipy.helpers.interceptors.grpc.metric.server_interceptor.AsyncGrpcServerMetricInterceptor.ONE_TO_FIVE_SECONDS_BUCKETS class-attribute

ONE_TO_FIVE_SECONDS_BUCKETS: list[float] = [
    (i / 100) for i in (range(100, 500, 20))
]

Buckets for measuring response times between 5 and 30 seconds.

archipy.helpers.interceptors.grpc.metric.server_interceptor.AsyncGrpcServerMetricInterceptor.FIVE_TO_THIRTY_SECONDS_BUCKETS class-attribute

FIVE_TO_THIRTY_SECONDS_BUCKETS: list[float] = [
    (i / 100) for i in (range(500, 3000, 50))
]

Combined buckets for measuring response times from 0 to 30 seconds and beyond.

archipy.helpers.interceptors.grpc.metric.server_interceptor.AsyncGrpcServerMetricInterceptor.TOTAL_BUCKETS class-attribute instance-attribute

TOTAL_BUCKETS = (
    ZERO_TO_ONE_SECONDS_BUCKETS
    + ONE_TO_FIVE_SECONDS_BUCKETS
    + FIVE_TO_THIRTY_SECONDS_BUCKETS
    + [float("inf")]
)

Prometheus histogram for tracking response times of async gRPC methods.

archipy.helpers.interceptors.grpc.metric.server_interceptor.AsyncGrpcServerMetricInterceptor.RESPONSE_TIME_SECONDS class-attribute instance-attribute

RESPONSE_TIME_SECONDS = Histogram(
    "grpc_async_response_time_seconds",
    "Time spent processing async gRPC request",
    labelnames=(
        "package",
        "service",
        "method",
        "status_code",
    ),
    buckets=TOTAL_BUCKETS,
)

Prometheus gauge for tracking active async gRPC requests.

archipy.helpers.interceptors.grpc.metric.server_interceptor.AsyncGrpcServerMetricInterceptor.ACTIVE_REQUESTS class-attribute instance-attribute

ACTIVE_REQUESTS = Gauge(
    "grpc_async_active_requests",
    "Number of active async gRPC requests",
    labelnames=("package", "service", "method"),
)

archipy.helpers.interceptors.grpc.metric.server_interceptor.AsyncGrpcServerMetricInterceptor.intercept async

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

Intercepts an async gRPC server call to measure response time and track active requests.

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.

Raises:

Type	Description
Exception	If an exception occurs during the method execution, it is captured and logged.

Source code in archipy/helpers/interceptors/grpc/metric/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 to measure response time and track active requests.

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

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

    Raises:
        Exception: If an exception occurs during the method execution, it is captured and logged.
    """
    if not BaseConfig.global_config().PROMETHEUS.IS_ENABLED:
        return await method(request, context)

    package = method_name_model.package
    service = method_name_model.service
    method_name = method_name_model.method

    self.ACTIVE_REQUESTS.labels(package=package, service=service, method=method_name).inc()

    start_time = asyncio.get_event_loop().time()
    status_code = "OK"

    try:
        try:
            result = await method(request, context)

            if hasattr(context, "code") and context.code():
                status_code = context.code().name
        except Exception as exception:
            if isinstance(exception, grpc.aio.AioRpcError):
                code_obj = exception.code()
                if code_obj is not None:
                    code_name = getattr(code_obj, "name", None)
                    if code_name is not None:
                        status_code = code_name
            elif hasattr(exception, "code") and callable(exception.code):
                code_method = exception.code
                code_obj = code_method()  # ty: ignore[call-top-callable]
                if code_obj is not None:
                    code_name = getattr(code_obj, "name", None)
                    if code_name is not None:
                        status_code = code_name
            else:
                status_code = "INTERNAL"
            raise
        else:
            return result
        finally:
            duration = asyncio.get_event_loop().time() - start_time
            self.RESPONSE_TIME_SECONDS.labels(
                package=package,
                service=service,
                method=method_name,
                status_code=status_code,
            ).observe(duration)
            self.ACTIVE_REQUESTS.labels(package=package, service=service, method=method_name).dec()

    except Exception as exception:
        BaseUtils.capture_exception(exception)
        raise

archipy.helpers.interceptors.grpc.metric.server_interceptor.AsyncGrpcServerMetricInterceptor.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

trace

gRPC interceptors for propagating distributed tracing context across client and server.

gRPC client trace interceptors (Elastic APM + Sentry propagation).

archipy.helpers.interceptors.grpc.trace.client_interceptor.logger module-attribute

logger = logging.getLogger(__name__)

archipy.helpers.interceptors.grpc.trace.client_interceptor.GrpcClientTraceInterceptor

Bases: BaseGrpcClientInterceptor

gRPC client interceptor: Elastic APM external span + W3C and Sentry trace headers.

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

class GrpcClientTraceInterceptor(BaseGrpcClientInterceptor):
    """gRPC client interceptor: Elastic APM external span + W3C and Sentry trace headers."""

    def intercept(self, method: Callable, request_or_iterator: Any, call_details: grpc.ClientCallDetails) -> Any:
        """Intercept a unary-style client call with full APM span and trace propagation."""
        config = BaseConfig.global_config()
        if not TracingUtils.is_tracing_enabled(config):
            return method(request_or_iterator, call_details)

        TracingUtils.init_tracing_if_needed(config)

        if not config.ELASTIC_APM.IS_ENABLED:
            return self._intercept_sentry_headers_only(method, request_or_iterator, call_details, config)

        try:
            import elasticapm
            from elasticapm.traces import DroppedSpan
        except ImportError:
            logger.debug("elasticapm is not installed, using header-only tracing for gRPC client.")
            return self._intercept_sentry_headers_only(method, request_or_iterator, call_details, config)

        extra = _destination_extra(call_details.method)
        with elasticapm.capture_span(
            call_details.method,
            span_type="external",
            span_subtype="grpc",
            extra=extra,
            leaf=True,
        ) as apm_span:
            if not apm_span or isinstance(apm_span, DroppedSpan):
                return self._intercept_sentry_headers_only(method, request_or_iterator, call_details, config)
            metadata = TracingUtils.inject_outbound_metadata(
                call_details.metadata,
                apm_span,
                include_sentry=config.SENTRY.IS_ENABLED,
            )
            new_details = ClientCallDetails(
                method=call_details.method,
                timeout=call_details.timeout,
                metadata=metadata,
                credentials=call_details.credentials,
                wait_for_ready=call_details.wait_for_ready,
                compression=call_details.compression,
            )
            try:
                return method(request_or_iterator, new_details)
            except grpc.RpcError:
                apm_span.set_failure()
                raise

    def _intercept_sentry_headers_only(
        self,
        method: Callable,
        request_or_iterator: Any,
        call_details: grpc.ClientCallDetails,
        config: BaseConfig,
    ) -> Any:
        metadata = TracingUtils.inject_outbound_metadata(
            call_details.metadata,
            None,
            include_sentry=config.SENTRY.IS_ENABLED,
        )
        new_details = ClientCallDetails(
            method=call_details.method,
            timeout=call_details.timeout,
            metadata=metadata,
            credentials=call_details.credentials,
            wait_for_ready=call_details.wait_for_ready,
            compression=call_details.compression,
        )
        return method(request_or_iterator, new_details)

    def intercept_unary_stream(
        self,
        continuation: Callable[[grpc.ClientCallDetails, _TRequest], Any],
        client_call_details: grpc.ClientCallDetails,
        request: _TRequest,
    ) -> Any:
        """Propagate trace headers only (no APM span) for unary-stream RPCs."""
        return self._intercept_streaming(_swap_args(continuation), request, client_call_details)

    def intercept_stream_unary(
        self,
        continuation: Callable[[grpc.ClientCallDetails, Any], Any],
        client_call_details: grpc.ClientCallDetails,
        request_iterator: Any,
    ) -> Any:
        """Propagate trace headers only (no APM span) for stream-unary RPCs."""
        return self._intercept_streaming(_swap_args(continuation), request_iterator, client_call_details)

    def intercept_stream_stream(
        self,
        continuation: Callable[[grpc.ClientCallDetails, Any], Any],
        client_call_details: grpc.ClientCallDetails,
        request_iterator: Any,
    ) -> Any:
        """Propagate trace headers only (no APM span) for stream-stream RPCs."""
        return self._intercept_streaming(_swap_args(continuation), request_iterator, client_call_details)

    def _intercept_streaming(
        self,
        method: Callable,
        request_or_iterator: Any,
        call_details: grpc.ClientCallDetails,
    ) -> Any:
        config = BaseConfig.global_config()
        if not TracingUtils.is_tracing_enabled(config):
            return method(request_or_iterator, call_details)
        TracingUtils.init_tracing_if_needed(config)
        metadata = _streaming_client_metadata(call_details, config)
        new_details = ClientCallDetails(
            method=call_details.method,
            timeout=call_details.timeout,
            metadata=metadata,
            credentials=call_details.credentials,
            wait_for_ready=call_details.wait_for_ready,
            compression=call_details.compression,
        )
        return method(request_or_iterator, new_details)

archipy.helpers.interceptors.grpc.trace.client_interceptor.GrpcClientTraceInterceptor.intercept

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

Intercept a unary-style client call with full APM span and trace propagation.

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

def intercept(self, method: Callable, request_or_iterator: Any, call_details: grpc.ClientCallDetails) -> Any:
    """Intercept a unary-style client call with full APM span and trace propagation."""
    config = BaseConfig.global_config()
    if not TracingUtils.is_tracing_enabled(config):
        return method(request_or_iterator, call_details)

    TracingUtils.init_tracing_if_needed(config)

    if not config.ELASTIC_APM.IS_ENABLED:
        return self._intercept_sentry_headers_only(method, request_or_iterator, call_details, config)

    try:
        import elasticapm
        from elasticapm.traces import DroppedSpan
    except ImportError:
        logger.debug("elasticapm is not installed, using header-only tracing for gRPC client.")
        return self._intercept_sentry_headers_only(method, request_or_iterator, call_details, config)

    extra = _destination_extra(call_details.method)
    with elasticapm.capture_span(
        call_details.method,
        span_type="external",
        span_subtype="grpc",
        extra=extra,
        leaf=True,
    ) as apm_span:
        if not apm_span or isinstance(apm_span, DroppedSpan):
            return self._intercept_sentry_headers_only(method, request_or_iterator, call_details, config)
        metadata = TracingUtils.inject_outbound_metadata(
            call_details.metadata,
            apm_span,
            include_sentry=config.SENTRY.IS_ENABLED,
        )
        new_details = ClientCallDetails(
            method=call_details.method,
            timeout=call_details.timeout,
            metadata=metadata,
            credentials=call_details.credentials,
            wait_for_ready=call_details.wait_for_ready,
            compression=call_details.compression,
        )
        try:
            return method(request_or_iterator, new_details)
        except grpc.RpcError:
            apm_span.set_failure()
            raise

archipy.helpers.interceptors.grpc.trace.client_interceptor.GrpcClientTraceInterceptor.intercept_unary_stream

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

Propagate trace headers only (no APM span) for unary-stream RPCs.

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

def intercept_unary_stream(
    self,
    continuation: Callable[[grpc.ClientCallDetails, _TRequest], Any],
    client_call_details: grpc.ClientCallDetails,
    request: _TRequest,
) -> Any:
    """Propagate trace headers only (no APM span) for unary-stream RPCs."""
    return self._intercept_streaming(_swap_args(continuation), request, client_call_details)

archipy.helpers.interceptors.grpc.trace.client_interceptor.GrpcClientTraceInterceptor.intercept_stream_unary

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

Propagate trace headers only (no APM span) for stream-unary RPCs.

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

def intercept_stream_unary(
    self,
    continuation: Callable[[grpc.ClientCallDetails, Any], Any],
    client_call_details: grpc.ClientCallDetails,
    request_iterator: Any,
) -> Any:
    """Propagate trace headers only (no APM span) for stream-unary RPCs."""
    return self._intercept_streaming(_swap_args(continuation), request_iterator, client_call_details)

archipy.helpers.interceptors.grpc.trace.client_interceptor.GrpcClientTraceInterceptor.intercept_stream_stream

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

Propagate trace headers only (no APM span) for stream-stream RPCs.

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

def intercept_stream_stream(
    self,
    continuation: Callable[[grpc.ClientCallDetails, Any], Any],
    client_call_details: grpc.ClientCallDetails,
    request_iterator: Any,
) -> Any:
    """Propagate trace headers only (no APM span) for stream-stream RPCs."""
    return self._intercept_streaming(_swap_args(continuation), request_iterator, client_call_details)

archipy.helpers.interceptors.grpc.trace.client_interceptor.GrpcClientTraceInterceptor.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.
    """
    result = self.intercept(_swap_args(continuation), request, client_call_details)
    return result

archipy.helpers.interceptors.grpc.trace.client_interceptor.AsyncGrpcClientTraceInterceptor

Bases: BaseAsyncGrpcClientInterceptor

Async gRPC client interceptor: Elastic APM external span + W3C and Sentry trace headers.

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

class AsyncGrpcClientTraceInterceptor(BaseAsyncGrpcClientInterceptor):
    """Async gRPC client interceptor: Elastic APM external span + W3C and Sentry trace headers."""

    async def intercept(
        self,
        method: Callable,
        request_or_iterator: Any,
        call_details: grpc.aio.ClientCallDetails,
    ) -> Any:
        """Intercept an async unary-style client call with full APM span and trace propagation."""
        config = BaseConfig.global_config()
        if not TracingUtils.is_tracing_enabled(config):
            return await method(request_or_iterator, call_details)

        TracingUtils.init_tracing_if_needed(config)

        if not config.ELASTIC_APM.IS_ENABLED:
            return await self._intercept_sentry_headers_only_async(method, request_or_iterator, call_details, config)

        try:
            import elasticapm
            from elasticapm.traces import DroppedSpan
        except ImportError:
            logger.debug("elasticapm is not installed, using header-only tracing for async gRPC client.")
            return await self._intercept_sentry_headers_only_async(method, request_or_iterator, call_details, config)

        extra = _destination_extra(call_details.method)
        async with elasticapm.async_capture_span(
            call_details.method,
            span_type="external",
            span_subtype="grpc",
            extra=extra,
            leaf=True,
        ) as apm_span:
            if not apm_span or isinstance(apm_span, DroppedSpan):
                return await self._intercept_sentry_headers_only_async(
                    method,
                    request_or_iterator,
                    call_details,
                    config,
                )
            metadata = TracingUtils.inject_outbound_metadata(
                call_details.metadata,
                apm_span,
                include_sentry=config.SENTRY.IS_ENABLED,
            )
            new_details = AsyncClientCallDetails(
                method=call_details.method,
                timeout=call_details.timeout,
                metadata=metadata,
                credentials=call_details.credentials,
                wait_for_ready=call_details.wait_for_ready,
            )
            try:
                return await method(request_or_iterator, new_details)
            except grpc.RpcError:
                apm_span.set_failure()
                raise

    async def _intercept_sentry_headers_only_async(
        self,
        method: Callable,
        request_or_iterator: Any,
        call_details: grpc.aio.ClientCallDetails,
        config: BaseConfig,
    ) -> Any:
        metadata = TracingUtils.inject_outbound_metadata(
            call_details.metadata,
            None,
            include_sentry=config.SENTRY.IS_ENABLED,
        )
        new_details = AsyncClientCallDetails(
            method=call_details.method,
            timeout=call_details.timeout,
            metadata=metadata,
            credentials=call_details.credentials,
            wait_for_ready=call_details.wait_for_ready,
        )
        return await method(request_or_iterator, new_details)

    async def intercept_unary_stream(
        self,
        continuation: Callable[[grpc.aio.ClientCallDetails, _TRequest], Any],
        client_call_details: grpc.aio.ClientCallDetails,
        request: _TRequest,
    ) -> Any:
        """Propagate trace headers only (no APM span) for async unary-stream RPCs."""
        return await self._intercept_streaming_async(_swap_args(continuation), request, client_call_details)

    async def intercept_stream_unary(
        self,
        continuation: Callable[[grpc.aio.ClientCallDetails, Any], Any],
        client_call_details: grpc.aio.ClientCallDetails,
        request_iterator: Any,
    ) -> Any:
        """Propagate trace headers only (no APM span) for async stream-unary RPCs."""
        return await self._intercept_streaming_async(_swap_args(continuation), request_iterator, client_call_details)

    async def intercept_stream_stream(
        self,
        continuation: Callable[[grpc.aio.ClientCallDetails, Any], Any],
        client_call_details: grpc.aio.ClientCallDetails,
        request_iterator: Any,
    ) -> Any:
        """Propagate trace headers only (no APM span) for async stream-stream RPCs."""
        return await self._intercept_streaming_async(_swap_args(continuation), request_iterator, client_call_details)

    async def _intercept_streaming_async(
        self,
        method: Callable,
        request_or_iterator: Any,
        call_details: grpc.aio.ClientCallDetails,
    ) -> Any:
        config = BaseConfig.global_config()
        if not TracingUtils.is_tracing_enabled(config):
            return await method(request_or_iterator, call_details)
        TracingUtils.init_tracing_if_needed(config)
        metadata = _streaming_client_metadata(call_details, config)
        new_details = AsyncClientCallDetails(
            method=call_details.method,
            timeout=call_details.timeout,
            metadata=metadata,
            credentials=call_details.credentials,
            wait_for_ready=call_details.wait_for_ready,
        )
        return await method(request_or_iterator, new_details)

archipy.helpers.interceptors.grpc.trace.client_interceptor.AsyncGrpcClientTraceInterceptor.intercept async

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

Intercept an async unary-style client call with full APM span and trace propagation.

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

async def intercept(
    self,
    method: Callable,
    request_or_iterator: Any,
    call_details: grpc.aio.ClientCallDetails,
) -> Any:
    """Intercept an async unary-style client call with full APM span and trace propagation."""
    config = BaseConfig.global_config()
    if not TracingUtils.is_tracing_enabled(config):
        return await method(request_or_iterator, call_details)

    TracingUtils.init_tracing_if_needed(config)

    if not config.ELASTIC_APM.IS_ENABLED:
        return await self._intercept_sentry_headers_only_async(method, request_or_iterator, call_details, config)

    try:
        import elasticapm
        from elasticapm.traces import DroppedSpan
    except ImportError:
        logger.debug("elasticapm is not installed, using header-only tracing for async gRPC client.")
        return await self._intercept_sentry_headers_only_async(method, request_or_iterator, call_details, config)

    extra = _destination_extra(call_details.method)
    async with elasticapm.async_capture_span(
        call_details.method,
        span_type="external",
        span_subtype="grpc",
        extra=extra,
        leaf=True,
    ) as apm_span:
        if not apm_span or isinstance(apm_span, DroppedSpan):
            return await self._intercept_sentry_headers_only_async(
                method,
                request_or_iterator,
                call_details,
                config,
            )
        metadata = TracingUtils.inject_outbound_metadata(
            call_details.metadata,
            apm_span,
            include_sentry=config.SENTRY.IS_ENABLED,
        )
        new_details = AsyncClientCallDetails(
            method=call_details.method,
            timeout=call_details.timeout,
            metadata=metadata,
            credentials=call_details.credentials,
            wait_for_ready=call_details.wait_for_ready,
        )
        try:
            return await method(request_or_iterator, new_details)
        except grpc.RpcError:
            apm_span.set_failure()
            raise

archipy.helpers.interceptors.grpc.trace.client_interceptor.AsyncGrpcClientTraceInterceptor.intercept_unary_stream async

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

Propagate trace headers only (no APM span) for async unary-stream RPCs.

Source code in archipy/helpers/interceptors/grpc/trace/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:
    """Propagate trace headers only (no APM span) for async unary-stream RPCs."""
    return await self._intercept_streaming_async(_swap_args(continuation), request, client_call_details)

archipy.helpers.interceptors.grpc.trace.client_interceptor.AsyncGrpcClientTraceInterceptor.intercept_stream_unary async

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

Propagate trace headers only (no APM span) for async stream-unary RPCs.

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

async def intercept_stream_unary(
    self,
    continuation: Callable[[grpc.aio.ClientCallDetails, Any], Any],
    client_call_details: grpc.aio.ClientCallDetails,
    request_iterator: Any,
) -> Any:
    """Propagate trace headers only (no APM span) for async stream-unary RPCs."""
    return await self._intercept_streaming_async(_swap_args(continuation), request_iterator, client_call_details)

archipy.helpers.interceptors.grpc.trace.client_interceptor.AsyncGrpcClientTraceInterceptor.intercept_stream_stream async

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

Propagate trace headers only (no APM span) for async stream-stream RPCs.

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

async def intercept_stream_stream(
    self,
    continuation: Callable[[grpc.aio.ClientCallDetails, Any], Any],
    client_call_details: grpc.aio.ClientCallDetails,
    request_iterator: Any,
) -> Any:
    """Propagate trace headers only (no APM span) for async stream-stream RPCs."""
    return await self._intercept_streaming_async(_swap_args(continuation), request_iterator, client_call_details)

archipy.helpers.interceptors.grpc.trace.client_interceptor.AsyncGrpcClientTraceInterceptor.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.
    """
    result = await self.intercept(_swap_args(continuation), request, client_call_details)
    return result

options: show_root_toc_entry: false heading_level: 3

gRPC server trace interceptors (Elastic APM + Sentry).

archipy.helpers.interceptors.grpc.trace.server_interceptor.logger module-attribute

logger = logging.getLogger(__name__)

archipy.helpers.interceptors.grpc.trace.server_interceptor.GrpcServerTraceInterceptor

Bases: BaseGrpcServerInterceptor

Sync gRPC server interceptor: unary-unary only; Elastic transaction + Sentry trace continuation.

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

class GrpcServerTraceInterceptor(BaseGrpcServerInterceptor):
    """Sync gRPC server interceptor: unary-unary only; Elastic transaction + Sentry trace continuation."""

    def intercept_service(
        self,
        continuation: Callable[[grpc.HandlerCallDetails], grpc.RpcMethodHandler | None],
        handler_call_details: grpc.HandlerCallDetails,
    ) -> grpc.RpcMethodHandler | None:
        """Skip tracing for streaming RPCs (matches ``elasticapm`` gRPC server interceptor)."""
        next_handler = continuation(handler_call_details)
        if next_handler is None:
            return None
        if next_handler.request_streaming or next_handler.response_streaming:
            return next_handler

        handler_factory, next_handler_method = _get_factory_and_method(next_handler)

        def invoke_intercept_method(request: object, context: grpc.ServicerContext) -> object:
            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,
        )

    def intercept(
        self,
        method: Callable,
        request: object,
        context: grpc.ServicerContext,
        method_name_model: MethodName,
    ) -> object:
        """Run unary-unary handler inside Elastic and Sentry transactions."""
        config = BaseConfig.global_config()
        if not TracingUtils.is_tracing_enabled(config):
            return method(request, context)

        TracingUtils.init_tracing_if_needed(config)

        metadata_items = list(context.invocation_metadata())
        metadata_dict = _invocation_metadata_to_dict(metadata_items)

        sentry_transaction = None
        if config.SENTRY.IS_ENABLED:
            try:
                import sentry_sdk

                _, sentry_headers = TracingUtils.extract_inbound_trace(metadata_dict)
                sentry_transaction = sentry_sdk.continue_trace(
                    sentry_headers,
                    op="grpc.server",
                    name=method_name_model.full_name,
                )
                sentry_transaction.__enter__()
            except ImportError:
                logger.debug("sentry_sdk is not installed, skipping Sentry transaction for gRPC server.")
            except Exception:
                logger.exception("Failed to start Sentry transaction for gRPC server call")

        elastic_client: Any = None
        if config.ELASTIC_APM.IS_ENABLED:
            try:
                import elasticapm

                elastic_client = elasticapm.get_client()
                if elastic_client is None:
                    logger.warning("Elastic APM client is not initialized; skipping APM transaction for gRPC.")
                elif (parent := elasticapm.trace_parent_from_headers(metadata_dict)) is not None:
                    elastic_client.begin_transaction(transaction_type="request", trace_parent=parent)
                else:
                    elastic_client.begin_transaction(transaction_type="request")
            except ImportError:
                logger.debug("elasticapm is not installed, skipping Elastic APM transaction for gRPC server.")
            except Exception:
                logger.exception("Failed to begin Elastic APM transaction for gRPC server call")
                elastic_client = None

        wrapped_ctx = _ServicerContextWrapper(context, sentry_transaction)

        exc_info: tuple[type[BaseException] | None, BaseException | None, Any] = (None, None, None)
        try:
            result = method(request, wrapped_ctx)
        except Exception as exception:
            exc_info = (type(exception), exception, exception.__traceback__)
            if sentry_transaction is not None:
                sentry_transaction.set_status("internal_error")
            if elastic_client is not None:
                try:
                    import elasticapm

                    elasticapm.set_transaction_outcome(TracingUtils.outcome_for_exception(exception))
                except ImportError:
                    pass
                elastic_client.end_transaction(name=method_name_model.full_name, result="failure")
            raise
        else:
            try:
                import elasticapm
                from elasticapm.conf.constants import OUTCOME

                elasticapm.set_transaction_outcome(OUTCOME.SUCCESS, override=False)
            except ImportError:
                pass
            if sentry_transaction is not None and sentry_transaction.status is None:
                sentry_transaction.set_status("ok")
            if elastic_client is not None:
                elastic_client.end_transaction(name=method_name_model.full_name, result="success")
            return result
        finally:
            if sentry_transaction is not None:
                try:
                    sentry_transaction.__exit__(exc_info[0], exc_info[1], exc_info[2])
                except Exception:
                    logger.exception("Error closing Sentry transaction for gRPC server call")

archipy.helpers.interceptors.grpc.trace.server_interceptor.GrpcServerTraceInterceptor.intercept_service

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

Skip tracing for streaming RPCs (matches elasticapm gRPC server interceptor).

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

def intercept_service(
    self,
    continuation: Callable[[grpc.HandlerCallDetails], grpc.RpcMethodHandler | None],
    handler_call_details: grpc.HandlerCallDetails,
) -> grpc.RpcMethodHandler | None:
    """Skip tracing for streaming RPCs (matches ``elasticapm`` gRPC server interceptor)."""
    next_handler = continuation(handler_call_details)
    if next_handler is None:
        return None
    if next_handler.request_streaming or next_handler.response_streaming:
        return next_handler

    handler_factory, next_handler_method = _get_factory_and_method(next_handler)

    def invoke_intercept_method(request: object, context: grpc.ServicerContext) -> object:
        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.trace.server_interceptor.GrpcServerTraceInterceptor.intercept

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

Run unary-unary handler inside Elastic and Sentry transactions.

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

def intercept(
    self,
    method: Callable,
    request: object,
    context: grpc.ServicerContext,
    method_name_model: MethodName,
) -> object:
    """Run unary-unary handler inside Elastic and Sentry transactions."""
    config = BaseConfig.global_config()
    if not TracingUtils.is_tracing_enabled(config):
        return method(request, context)

    TracingUtils.init_tracing_if_needed(config)

    metadata_items = list(context.invocation_metadata())
    metadata_dict = _invocation_metadata_to_dict(metadata_items)

    sentry_transaction = None
    if config.SENTRY.IS_ENABLED:
        try:
            import sentry_sdk

            _, sentry_headers = TracingUtils.extract_inbound_trace(metadata_dict)
            sentry_transaction = sentry_sdk.continue_trace(
                sentry_headers,
                op="grpc.server",
                name=method_name_model.full_name,
            )
            sentry_transaction.__enter__()
        except ImportError:
            logger.debug("sentry_sdk is not installed, skipping Sentry transaction for gRPC server.")
        except Exception:
            logger.exception("Failed to start Sentry transaction for gRPC server call")

    elastic_client: Any = None
    if config.ELASTIC_APM.IS_ENABLED:
        try:
            import elasticapm

            elastic_client = elasticapm.get_client()
            if elastic_client is None:
                logger.warning("Elastic APM client is not initialized; skipping APM transaction for gRPC.")
            elif (parent := elasticapm.trace_parent_from_headers(metadata_dict)) is not None:
                elastic_client.begin_transaction(transaction_type="request", trace_parent=parent)
            else:
                elastic_client.begin_transaction(transaction_type="request")
        except ImportError:
            logger.debug("elasticapm is not installed, skipping Elastic APM transaction for gRPC server.")
        except Exception:
            logger.exception("Failed to begin Elastic APM transaction for gRPC server call")
            elastic_client = None

    wrapped_ctx = _ServicerContextWrapper(context, sentry_transaction)

    exc_info: tuple[type[BaseException] | None, BaseException | None, Any] = (None, None, None)
    try:
        result = method(request, wrapped_ctx)
    except Exception as exception:
        exc_info = (type(exception), exception, exception.__traceback__)
        if sentry_transaction is not None:
            sentry_transaction.set_status("internal_error")
        if elastic_client is not None:
            try:
                import elasticapm

                elasticapm.set_transaction_outcome(TracingUtils.outcome_for_exception(exception))
            except ImportError:
                pass
            elastic_client.end_transaction(name=method_name_model.full_name, result="failure")
        raise
    else:
        try:
            import elasticapm
            from elasticapm.conf.constants import OUTCOME

            elasticapm.set_transaction_outcome(OUTCOME.SUCCESS, override=False)
        except ImportError:
            pass
        if sentry_transaction is not None and sentry_transaction.status is None:
            sentry_transaction.set_status("ok")
        if elastic_client is not None:
            elastic_client.end_transaction(name=method_name_model.full_name, result="success")
        return result
    finally:
        if sentry_transaction is not None:
            try:
                sentry_transaction.__exit__(exc_info[0], exc_info[1], exc_info[2])
            except Exception:
                logger.exception("Error closing Sentry transaction for gRPC server call")

archipy.helpers.interceptors.grpc.trace.server_interceptor.AsyncGrpcServerTraceInterceptor

Bases: BaseAsyncGrpcServerInterceptor

Async gRPC server interceptor: unary-unary only; Elastic transaction + Sentry trace continuation.

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

class AsyncGrpcServerTraceInterceptor(BaseAsyncGrpcServerInterceptor):
    """Async gRPC server interceptor: unary-unary only; Elastic transaction + Sentry trace continuation."""

    async def intercept_service(
        self,
        continuation: Callable[[grpc.HandlerCallDetails], Awaitable[grpc.RpcMethodHandler | None]],
        handler_call_details: grpc.HandlerCallDetails,
    ) -> grpc.RpcMethodHandler | None:
        """Skip tracing for streaming RPCs (matches ``elasticapm`` async gRPC server interceptor)."""
        next_handler = await continuation(handler_call_details)
        if next_handler is None:
            return None
        if next_handler.request_streaming or next_handler.response_streaming:
            return next_handler

        handler_factory, next_handler_method = _get_factory_and_method(next_handler)

        async def invoke_intercept_method(request: object, context: grpc.aio.ServicerContext) -> object:
            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),
        )

    async def intercept(
        self,
        method: Callable,
        request: object,
        context: grpc.aio.ServicerContext,
        method_name_model: MethodName,
    ) -> object:
        """Run async unary-unary handler inside Elastic and Sentry transactions."""
        config = BaseConfig.global_config()
        if not TracingUtils.is_tracing_enabled(config):
            return await method(request, context)

        TracingUtils.init_tracing_if_needed(config)

        invocation_metadata = context.invocation_metadata()
        if invocation_metadata is not None:
            metadata_items = list(invocation_metadata)
        else:
            metadata_items = []
        metadata_dict = _invocation_metadata_to_dict(metadata_items)

        sentry_transaction = None
        if config.SENTRY.IS_ENABLED:
            try:
                import sentry_sdk

                _, sentry_headers = TracingUtils.extract_inbound_trace(metadata_dict)
                sentry_transaction = sentry_sdk.continue_trace(
                    sentry_headers,
                    op="grpc.server",
                    name=method_name_model.full_name,
                )
                sentry_transaction.__enter__()
            except ImportError:
                logger.debug("sentry_sdk is not installed, skipping Sentry transaction for async gRPC server.")
            except Exception:
                logger.exception("Failed to start Sentry transaction for async gRPC server call")

        elastic_client: Any = None
        if config.ELASTIC_APM.IS_ENABLED:
            try:
                import elasticapm

                elastic_client = elasticapm.get_client()
                if elastic_client is None:
                    logger.warning("Elastic APM client is not initialized; skipping APM transaction for gRPC.")
                elif (parent := elasticapm.trace_parent_from_headers(metadata_dict)) is not None:
                    elastic_client.begin_transaction(transaction_type="request", trace_parent=parent)
                else:
                    elastic_client.begin_transaction(transaction_type="request")
            except ImportError:
                logger.debug("elasticapm is not installed, skipping Elastic APM transaction for async gRPC server.")
            except Exception:
                logger.exception("Failed to begin Elastic APM transaction for async gRPC server call")
                elastic_client = None

        wrapped_ctx = _AsyncServicerContextWrapper(context, sentry_transaction)

        exc_info: tuple[type[BaseException] | None, BaseException | None, Any] = (None, None, None)
        try:
            result = method(request, wrapped_ctx)
            if inspect.isawaitable(result):
                result = await result
        except Exception as exception:
            exc_info = (type(exception), exception, exception.__traceback__)
            if sentry_transaction is not None:
                sentry_transaction.set_status("internal_error")
            if elastic_client is not None:
                try:
                    import elasticapm

                    elasticapm.set_transaction_outcome(TracingUtils.outcome_for_exception(exception))
                except ImportError:
                    pass
                elastic_client.end_transaction(name=method_name_model.full_name, result="failure")
            raise
        else:
            try:
                import elasticapm
                from elasticapm.conf.constants import OUTCOME

                elasticapm.set_transaction_outcome(OUTCOME.SUCCESS, override=False)
            except ImportError:
                pass
            if sentry_transaction is not None and sentry_transaction.status is None:
                sentry_transaction.set_status("ok")
            if elastic_client is not None:
                elastic_client.end_transaction(name=method_name_model.full_name, result="success")
            return result
        finally:
            if sentry_transaction is not None:
                try:
                    sentry_transaction.__exit__(exc_info[0], exc_info[1], exc_info[2])
                except Exception:
                    logger.exception("Error closing Sentry transaction for async gRPC server call")

archipy.helpers.interceptors.grpc.trace.server_interceptor.AsyncGrpcServerTraceInterceptor.intercept_service async

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

Skip tracing for streaming RPCs (matches elasticapm async gRPC server interceptor).

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

async def intercept_service(
    self,
    continuation: Callable[[grpc.HandlerCallDetails], Awaitable[grpc.RpcMethodHandler | None]],
    handler_call_details: grpc.HandlerCallDetails,
) -> grpc.RpcMethodHandler | None:
    """Skip tracing for streaming RPCs (matches ``elasticapm`` async gRPC server interceptor)."""
    next_handler = await continuation(handler_call_details)
    if next_handler is None:
        return None
    if next_handler.request_streaming or next_handler.response_streaming:
        return next_handler

    handler_factory, next_handler_method = _get_factory_and_method(next_handler)

    async def invoke_intercept_method(request: object, context: grpc.aio.ServicerContext) -> object:
        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.trace.server_interceptor.AsyncGrpcServerTraceInterceptor.intercept async

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

Run async unary-unary handler inside Elastic and Sentry transactions.

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

async def intercept(
    self,
    method: Callable,
    request: object,
    context: grpc.aio.ServicerContext,
    method_name_model: MethodName,
) -> object:
    """Run async unary-unary handler inside Elastic and Sentry transactions."""
    config = BaseConfig.global_config()
    if not TracingUtils.is_tracing_enabled(config):
        return await method(request, context)

    TracingUtils.init_tracing_if_needed(config)

    invocation_metadata = context.invocation_metadata()
    if invocation_metadata is not None:
        metadata_items = list(invocation_metadata)
    else:
        metadata_items = []
    metadata_dict = _invocation_metadata_to_dict(metadata_items)

    sentry_transaction = None
    if config.SENTRY.IS_ENABLED:
        try:
            import sentry_sdk

            _, sentry_headers = TracingUtils.extract_inbound_trace(metadata_dict)
            sentry_transaction = sentry_sdk.continue_trace(
                sentry_headers,
                op="grpc.server",
                name=method_name_model.full_name,
            )
            sentry_transaction.__enter__()
        except ImportError:
            logger.debug("sentry_sdk is not installed, skipping Sentry transaction for async gRPC server.")
        except Exception:
            logger.exception("Failed to start Sentry transaction for async gRPC server call")

    elastic_client: Any = None
    if config.ELASTIC_APM.IS_ENABLED:
        try:
            import elasticapm

            elastic_client = elasticapm.get_client()
            if elastic_client is None:
                logger.warning("Elastic APM client is not initialized; skipping APM transaction for gRPC.")
            elif (parent := elasticapm.trace_parent_from_headers(metadata_dict)) is not None:
                elastic_client.begin_transaction(transaction_type="request", trace_parent=parent)
            else:
                elastic_client.begin_transaction(transaction_type="request")
        except ImportError:
            logger.debug("elasticapm is not installed, skipping Elastic APM transaction for async gRPC server.")
        except Exception:
            logger.exception("Failed to begin Elastic APM transaction for async gRPC server call")
            elastic_client = None

    wrapped_ctx = _AsyncServicerContextWrapper(context, sentry_transaction)

    exc_info: tuple[type[BaseException] | None, BaseException | None, Any] = (None, None, None)
    try:
        result = method(request, wrapped_ctx)
        if inspect.isawaitable(result):
            result = await result
    except Exception as exception:
        exc_info = (type(exception), exception, exception.__traceback__)
        if sentry_transaction is not None:
            sentry_transaction.set_status("internal_error")
        if elastic_client is not None:
            try:
                import elasticapm

                elasticapm.set_transaction_outcome(TracingUtils.outcome_for_exception(exception))
            except ImportError:
                pass
            elastic_client.end_transaction(name=method_name_model.full_name, result="failure")
        raise
    else:
        try:
            import elasticapm
            from elasticapm.conf.constants import OUTCOME

            elasticapm.set_transaction_outcome(OUTCOME.SUCCESS, override=False)
        except ImportError:
            pass
        if sentry_transaction is not None and sentry_transaction.status is None:
            sentry_transaction.set_status("ok")
        if elastic_client is not None:
            elastic_client.end_transaction(name=method_name_model.full_name, result="success")
        return result
    finally:
        if sentry_transaction is not None:
            try:
                sentry_transaction.__exit__(exc_info[0], exc_info[1], exc_info[2])
            except Exception:
                logger.exception("Error closing Sentry transaction for async gRPC server call")

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 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."""
        from redis.exceptions import RedisError  # noqa: PLC0415

        windows = RateLimitUtils.get_rate_limit_windows_from_callable(method)
        if not windows:
            result = method(request, context)
            if inspect.isawaitable(result):
                return await result
            return result

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

        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
                result = method(request, context)
                if inspect.isawaitable(result):
                    return await result
                return result

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

        result = method(request, context)
        if inspect.isawaitable(result):
            return await result
        return result

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."""
    from redis.exceptions import RedisError  # noqa: PLC0415

    windows = RateLimitUtils.get_rate_limit_windows_from_callable(method)
    if not windows:
        result = method(request, context)
        if inspect.isawaitable(result):
            return await result
        return result

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

    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
            result = method(request, context)
            if inspect.isawaitable(result):
                return await result
            return result

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

    result = method(request, context)
    if inspect.isawaitable(result):
        return await result
    return result

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) >= 2:
            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