"""Context variables for tracking request-specific data.""" from contextlib import contextmanager from contextvars import ContextVar, Token from typing import Generator, Literal, overload from . import constants from .errors import ERROR_CTX_USER_ID_NOT_SET ctx_user_id: ContextVar[str | None] = ContextVar('user_id', default=None) @overload def get_user_id(strict: Literal[True]) -> str: ... @overload def get_user_id(strict: Literal[False] | None = None) -> str | None: ... def get_user_id(strict: bool | None = None) -> str | None: """Get the current user ID from context. Args: strict: If `True`, raises error when `user_id` is `None`. If `False`, returns `None` silently. If `None` (default), uses `STRICT_CONTEXT_VALIDATION` setting. """ use_strict = constants.STRICT_CONTEXT_VALIDATION if strict is None else strict user_id = ctx_user_id.get() if user_id is None and use_strict: raise ValueError(ERROR_CTX_USER_ID_NOT_SET) return user_id @contextmanager def set_user_context(user_id: str | None) -> Generator[None, None, None]: """Context manager to temporarily set user ID. Usage: ```python with set_user_context('user_123'): model.soft_delete() # Automatically tracked to user_123 ``` """ token = ctx_user_id.set(user_id) try: yield finally: ctx_user_id.reset(token) def set_user_id(user_id: str | None) -> Token[str | None]: """Set the context user ID.""" return ctx_user_id.set(user_id)