The example in the docs have on_end method signature wrong. Here the correct example:
from contextvars import ContextVar, Token
from connectrpc.code import Code
from connectrpc.errors import ConnectError
from connectrpc.request import RequestContext
_auth_token = ContextVar["auth_token"]("current_auth_token")
class ServerAuthInterceptor:
def __init__(self, valid_tokens: list[str]):
self._valid_tokens = valid_tokens
async def on_start(self, ctx: RequestContext) -> Token["auth_token"]:
authorization = ctx.request_headers().get("authorization")
if not authorization or not authorization.startswith("Bearer "):
raise ConnectError(Code.UNAUTHENTICATED, message="")
token = authorization[len("Bearer ") :]
if token not in self._valid_tokens:
raise ConnectError(Code.PERMISSION_DENIED, message="")
return _auth_token.set(token)
async def on_end(
self, token: Token["auth_token"], ctx: RequestContext, error: Exception | None
):
_auth_token.reset(token)
The example in the docs have
on_endmethod signature wrong. Here the correct example: