43 lines
1.2 KiB
Python
43 lines
1.2 KiB
Python
|
"""Request-level caching for user principals (group_ids + role_id).
|
||
|
|
|
||
|
|
Avoids N+1 queries by loading principals once per request in deps.py
|
||
|
|
and reusing them in visibility.py and permission_resolver.py.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import uuid
|
||
|
|
from contextvars import ContextVar
|
||
|
|
from dataclasses import dataclass
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class UserPrincipals:
|
||
|
|
"""Cached user principals for the current request."""
|
||
|
|
user_id: uuid.UUID
|
||
|
|
tenant_id: uuid.UUID
|
||
|
|
group_ids: list[uuid.UUID]
|
||
|
|
role_id: uuid.UUID | None
|
||
|
|
|
||
|
|
|
||
|
|
# ContextVar — async-safe, isolated per request
|
||
|
|
_principals_ctx: ContextVar[UserPrincipals | None] = ContextVar(
|
||
|
|
"user_principals", default=None
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def set_principals(principals: UserPrincipals) -> None:
|
||
|
|
"""Set principals for the current request context."""
|
||
|
|
_principals_ctx.set(principals)
|
||
|
|
|
||
|
|
|
||
|
|
def get_principals() -> UserPrincipals | None:
|
||
|
|
"""Get principals for the current request context, or None if not set."""
|
||
|
|
return _principals_ctx.get()
|
||
|
|
|
||
|
|
|
||
|
|
def clear_principals() -> None:
|
||
|
|
"""Clear principals at end of request (optional — ContextVar is
|
||
|
|
isolated per task, but explicit cleanup is good practice)."""
|
||
|
|
_principals_ctx.set(None)
|