69 lines
2.3 KiB
Python
69 lines
2.3 KiB
Python
"""Request context utilities for correlation and logging."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from contextvars import ContextVar
|
|
|
|
_REQUEST_ID: ContextVar[str] = ContextVar("request_id", default="-")
|
|
_GITEA_USER_TOKEN: ContextVar[str | None] = ContextVar("gitea_user_token", default=None)
|
|
_GITEA_USER_LOGIN: ContextVar[str | None] = ContextVar("gitea_user_login", default=None)
|
|
_CLIENT_ADDRESS: ContextVar[str | None] = ContextVar("client_address", default=None)
|
|
_GITEA_USER_SCOPES: ContextVar[tuple[str, ...]] = ContextVar("gitea_user_scopes", default=())
|
|
|
|
|
|
def set_request_id(request_id: str) -> None:
|
|
"""Store request id in context-local state."""
|
|
_REQUEST_ID.set(request_id)
|
|
|
|
|
|
def get_request_id() -> str:
|
|
"""Get current request id from context-local state."""
|
|
return _REQUEST_ID.get()
|
|
|
|
|
|
def set_client_address(address: str | None) -> None:
|
|
"""Store the rate-limit bucket of the client address (set by the HTTP middleware)."""
|
|
_CLIENT_ADDRESS.set(address)
|
|
|
|
|
|
def get_client_address() -> str | None:
|
|
"""Rate-limit bucket of the current request's client address (None outside HTTP)."""
|
|
return _CLIENT_ADDRESS.get()
|
|
|
|
|
|
def set_gitea_user_token(token: str) -> None:
|
|
"""Store the per-request Gitea OAuth user token in context-local state."""
|
|
_GITEA_USER_TOKEN.set(token)
|
|
|
|
|
|
def get_gitea_user_token() -> str | None:
|
|
"""Get the per-request Gitea OAuth user token from context-local state."""
|
|
return _GITEA_USER_TOKEN.get()
|
|
|
|
|
|
def set_gitea_user_login(login: str) -> None:
|
|
"""Store the authenticated Gitea username in context-local state."""
|
|
_GITEA_USER_LOGIN.set(login)
|
|
|
|
|
|
def get_gitea_user_login() -> str | None:
|
|
"""Get the authenticated Gitea username from context-local state."""
|
|
return _GITEA_USER_LOGIN.get()
|
|
|
|
|
|
def set_gitea_user_scopes(scopes: list[str] | set[str] | tuple[str, ...]) -> None:
|
|
"""Store normalized OAuth scopes for the current request."""
|
|
_GITEA_USER_SCOPES.set(tuple(sorted({scope.strip() for scope in scopes if scope.strip()})))
|
|
|
|
|
|
def get_gitea_user_scopes() -> tuple[str, ...]:
|
|
"""Get OAuth scopes attached to the current request."""
|
|
return _GITEA_USER_SCOPES.get()
|
|
|
|
|
|
def clear_gitea_auth_context() -> None:
|
|
"""Reset per-request Gitea authentication context values."""
|
|
_GITEA_USER_TOKEN.set(None)
|
|
_GITEA_USER_LOGIN.set(None)
|
|
_GITEA_USER_SCOPES.set(())
|