Visitar URL original
Interceptors | ConnectSkip to content
Connect
Created by

Interceptors

Interceptors are similar to the middleware or decorators you may be familiar with from other frameworks: they’re the primary way of extending Connect. They can modify the context, the request, the response, and any errors. Interceptors are often used to add logging, metrics, tracing, retries, and other functionality.

Take care when writing interceptors! They’re powerful, but overly complex interceptors can make debugging difficult.

Interceptors are protocol implementations

Section titled “Interceptors are protocol implementations”

Connect interceptors are protocol implementations with the same signature as an RPC handler, along with a call_next Callable to continue with request processing. This allows writing interceptors in much the same way as any handler, making sure to call call_next when needing to call business logic - or not, if overriding the response within the interceptor itself.

Connect supports unary RPC and three stream types - because each has a different handler signature, we provide protocols corresponding to each.

class UnaryInterceptor(Protocol):


    async def intercept_unary(
        self,
        call_next: Callable[[REQ, RequestContext], Awaitable[RES]],
        request: REQ,
        ctx: RequestContext,
        /,
    ) -> RES: ...


class ClientStreamInterceptor(Protocol):


    async def intercept_client_stream(
        self,
        call_next: Callable[[AsyncIterator[REQ], RequestContext], Awaitable[RES]],
        request: AsyncIterator[REQ],
        ctx: RequestContext,
        /,
    ) -> RES: ...


class ServerStreamInterceptor(Protocol):


    def intercept_server_stream(
        self,
        call_next: Callable[[REQ, RequestContext], AsyncIterator[RES]],
        request: REQ,
        ctx: RequestContext,
        /,
    ) -> AsyncIterator[RES]: ...


class BidiStreamInterceptor(Protocol):


    def intercept_bidi_stream(
        self,
        call_next: Callable[[AsyncIterator[REQ], RequestContext], AsyncIterator[RES]],
        request: AsyncIterator[REQ],
        ctx: RequestContext,
        /,
    ) -> AsyncIterator[RES]: ...

A single class can implement as many of the protocols as needed.

An example

Section titled “An example”

That’s a little abstract, so let’s consider an example: we’d like to apply a filter to our greeting service from the getting started documentation that says “Goodbye” instead of “Hello” to certain callers.

from collections.abc import Awaitable, Callable


class GoodbyeInterceptor:
    def __init__(self, users: list[str]) -> None:
        self._users = users


    async def intercept_unary(
        self,
        call_next: Callable[[GreetRequest, RequestContext[GreetRequest, GreetResponse]], Awaitable[GreetResponse]],
        request: GreetRequest,
        ctx: RequestContext[GreetRequest, GreetResponse],
    ) -> GreetResponse:
        if request.name in self._users:
            return GreetResponse(greeting=f"Goodbye, {request.name}!")
        return await call_next(request, ctx)

To apply our new interceptor to handlers, we can pass it to the application with interceptors=.

app = GreetServiceASGIApplication(Greeter(), interceptors=[GoodbyeInterceptor(["user1", "user2"])])

Client constructors also accept an interceptors= parameter.

client = GreetServiceClient("http://localhost:8000", interceptors=[GoodbyeInterceptor(["user1", "user2"])])

Metadata interceptors

Section titled “Metadata interceptors”

Because the signature is different for each RPC type, we have an interceptor protocol for each to be able to intercept RPC messages. However, many interceptors, such as for metrics or tracing, only need access to headers and not messages. Connect provides a metadata interceptor protocol that can be implemented to work with any RPC type.

An interceptor timing each RPC and logging its duration may look like this:

import logging
import time


logger = logging.getLogger(__name__)


class TimingInterceptor:
    async def on_start(self, ctx: RequestContext) -> float:
        return time.perf_counter()


    async def on_end(self, start: float, ctx: RequestContext, error: Exception | None) -> None:
        duration = time.perf_counter() - start
        logger.info("%s took %.3fs", ctx.method.name, duration)

on_start can return any value, which is passed to the optional on_end method. Here, we return the start time to compute the RPC’s duration.

Authentication

Section titled “Authentication”

Don’t use interceptors to authenticate requests on the server. Handlers run unary interceptors after the request message has been read, decompressed, and deserialized. An interceptor-based check lets unauthenticated clients consume memory and CPU on your server. Instead, authenticate with ASGI or WSGI middleware, which runs before Connect reads the request body.

The middleware below checks the Authorization header and rejects failures with HTTP 401, which clients report as unauthenticated for all three protocols. It passes the caller’s identity to handlers in a ContextVar, since Connect handlers don’t see the ASGI scope or WSGI environ.

import hmac
from contextvars import ContextVar


current_user: ContextVar[str] = ContextVar("current_user")




def authenticate(authorization: str) -> str | None:
    # Look the credentials up in your user store. This example accepts one token.
    if hmac.compare_digest(authorization.encode(), b"Bearer open-sesame"):
        return "alice"
    return None




class AuthMiddleware:
    def __init__(self, app) -> None:
        self.app = app


    async def __call__(self, scope, receive, send) -> None:
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return
        headers = dict(scope["headers"])
        user = authenticate(headers.get(b"authorization", b"").decode("latin-1"))
        if user is None:
            await send({"type": "http.response.start", "status": 401, "headers": []})
            await send({"type": "http.response.body", "body": b""})
            return
        token = current_user.set(user)
        try:
            await self.app(scope, receive, send)
        finally:
            current_user.reset(token)

Wrap your application with the middleware, and read current_user in handlers:

class Greeter:
    async def greet(self, request: GreetRequest, ctx: RequestContext) -> GreetResponse:
        return GreetResponse(greeting=f"Hello, {request.name}! You are {current_user.get()}.")


app = AuthMiddleware(GreetServiceASGIApplication(Greeter()))