Coverage for src/lexigram/admin/state/context.py: 4%
131 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-24 23:31 +0800
1"""Request context for lexigram-admin.
3Provides a context manager for request-scoped state that integrates
4with lexigram-cache RequestContext for caching and with HTMX patterns.
5"""
7from __future__ import annotations
9from contextvars import ContextVar
10from dataclasses import dataclass, field
11from functools import wraps
12from typing import TYPE_CHECKING, Any, ClassVar, TypeVar
14from lexigram.logging import get_logger
16if TYPE_CHECKING:
17 from collections.abc import Awaitable, Callable
19 from starlette.requests import Request
21logger = get_logger(__name__)
23T = TypeVar("T")
26@dataclass
27class AdminContext:
28 """Request-scoped context for admin operations.
30 Holds all request-related state in a single immutable container.
31 This avoids mutation issues and makes state flow explicit.
33 Attributes:
34 request: The Starlette Request object
35 user: Authenticated user (or None)
36 permissions: User's permission set
37 resource: Current resource being accessed (if any)
38 action: Current action (list, view, create, update, delete)
39 entity_id: ID of entity being accessed (if any)
40 flash_messages: List of flash messages to display
41 cache_context: lexigram-cache request context
42 htmx: HTMX request info
43 breadcrumbs: Current breadcrumb trail
44 meta: Arbitrary metadata
45 """
47 request: Request
48 user: Any = None
49 permissions: Any = None
50 resource: str | None = None
51 action: str | None = None
52 entity_id: Any = None
53 flash_messages: list[dict[str, str]] = field(default_factory=list)
54 cache_context: Any = None
55 htmx: HTMXInfo | None = None
56 breadcrumbs: list[dict[str, str]] = field(default_factory=list)
57 meta: dict[str, Any] = field(default_factory=dict)
59 @property
60 def is_authenticated(self) -> bool:
61 """Check if user is authenticated."""
62 return self.user is not None
64 @property
65 def is_htmx(self) -> bool:
66 """Check if this request expects a fragment swap."""
67 if self.htmx is None or not self.htmx.is_htmx:
68 return False
69 return wants_fragment(self.request)
71 def can(self, permission: str) -> bool:
72 """Check if user has permission."""
73 if self.permissions is None:
74 return False
75 return self.permissions.has(permission)
77 def add_flash(self, message: str, category: str = "info") -> None:
78 """Add a flash message."""
79 self.flash_messages.append({"message": message, "category": category})
81 def add_breadcrumb(self, label: str, url: str | None = None) -> None:
82 """Add a breadcrumb."""
83 self.breadcrumbs.append({"label": label, "url": url}) # type: ignore[dict-item]
85 def with_resource(
86 self,
87 resource: str,
88 action: str,
89 entity_id: Any = None,
90 ) -> AdminContext:
91 """Create new context with resource info."""
92 return AdminContext(
93 request=self.request,
94 user=self.user,
95 permissions=self.permissions,
96 resource=resource,
97 action=action,
98 entity_id=entity_id,
99 flash_messages=list(self.flash_messages),
100 cache_context=self.cache_context,
101 htmx=self.htmx,
102 breadcrumbs=list(self.breadcrumbs),
103 meta=dict(self.meta),
104 )
107@dataclass
108class HTMXInfo:
109 """Information about HTMX request."""
111 is_htmx: bool = False
112 boosted: bool = False
113 target: str | None = None
114 trigger: str | None = None
115 trigger_name: str | None = None
116 prompt: str | None = None
117 history_restore: bool = False
119 @classmethod
120 def from_request(cls, request: Request) -> HTMXInfo:
121 """Extract HTMX info from request headers."""
122 return cls(
123 is_htmx=request.headers.get("HX-Request", "").lower() == "true",
124 boosted=request.headers.get("HX-Boosted", "").lower() == "true",
125 target=request.headers.get("HX-Target"),
126 trigger=request.headers.get("HX-Trigger"),
127 trigger_name=request.headers.get("HX-Trigger-Name"),
128 prompt=request.headers.get("HX-Prompt"),
129 history_restore=request.headers.get(
130 "HX-History-Restore-Request",
131 "",
132 ).lower()
133 == "true",
134 )
137def wants_fragment(request: Request) -> bool:
138 """Return True when the request expects a fragment swap.
140 Requests that carry an explicit ``HX-Target`` header (other than
141 ``body``) are fragment swaps. Boosted navigations and plain requests
142 carry no target and must receive a full page instead.
143 """
144 target = request.headers.get("HX-Target")
145 return bool(target and target != "body")
148class AdminContextManager:
149 """Context manager for admin request context.
151 Usage:
152 async with AdminContextManager(request) as ctx:
153 # ctx is AdminContext with user, permissions, etc.
154 pass
155 """
157 _context_var: ClassVar[ContextVar[AdminContext | None]] = ContextVar(
158 "admin_context",
159 default=None,
160 )
162 @classmethod
163 def get_context(cls) -> AdminContext | None:
164 """Get the current admin context for this async task."""
165 return cls._context_var.get()
167 @classmethod
168 def get_context_or_raise(cls) -> AdminContext:
169 """Get the current admin context or raise if not set."""
170 ctx = cls._context_var.get()
171 if ctx is None:
172 raise RuntimeError(
173 "No admin context set. Are you inside a request handler?"
174 )
175 return ctx
177 def __init__(self, request: Request):
178 self.request = request
179 self.ctx: AdminContext | None = None
180 self._token: Any = None
182 def _read_flash_from_session(self) -> list[dict[str, str]]:
183 """Read flash messages from session and clear them."""
184 try:
185 session = getattr(self.request, "session", None)
186 if session is None:
187 return []
188 return session.pop("_flash", [])
189 except Exception:
190 return []
192 def _write_flash_to_session(self, messages: list[dict[str, str]]) -> None:
193 """Write flash messages to session."""
194 try:
195 session = getattr(self.request, "session", None)
196 if session is None:
197 return
198 if messages:
199 session["_flash"] = messages
200 else:
201 session.pop("_flash", None)
202 except Exception: # noqa: S110 — intentional best-effort fallback
203 pass
205 async def __aenter__(self) -> AdminContext:
206 """Set up context for request."""
207 # Extract user from request.state (set by auth middleware)
208 user = getattr(self.request.state, "user", None)
209 permissions = getattr(self.request.state, "permissions", None)
211 # Create HTMX info
212 htmx = HTMXInfo.from_request(self.request)
214 # Create cache context if available
215 cache_ctx = None
216 try:
217 from lexigram.primitives.context import RequestContext
219 cache_ctx = RequestContext() # type: ignore[call-arg]
220 except (RuntimeError, ImportError, TypeError):
221 pass
223 # Create context
224 self.ctx = AdminContext(
225 request=self.request,
226 user=user,
227 permissions=permissions,
228 htmx=htmx,
229 cache_context=cache_ctx,
230 )
232 # Restore flash messages from session
233 self.ctx.flash_messages = self._read_flash_from_session()
235 # Set in context var
236 self._token = AdminContextManager._context_var.set(self.ctx)
238 return self.ctx
240 async def __aexit__(self, exc_type, exc_val, exc_tb) -> Any:
241 """Clean up context."""
242 # Persist any unconsumed flash messages back to session
243 if self.ctx:
244 self._write_flash_to_session(self.ctx.flash_messages)
245 AdminContextManager._context_var.reset(self._token)
246 self.ctx = None
247 return False
250def with_context(func: Callable[..., Awaitable[T]]) -> Callable[..., Awaitable[T]]:
251 """Decorator to automatically set up admin context.
253 Usage:
254 @with_context
255 async def my_handler(request: Request, ctx: AdminContext):
256 # ctx is automatically provided
257 pass
258 """
260 @wraps(func)
261 async def wrapper(request: Request, *args, **kwargs) -> Any:
262 async with AdminContextManager(request) as ctx:
263 return await func(request, ctx, *args, **kwargs)
265 return wrapper
268# Flash message helpers
269def flash(message: str, category: str = "info") -> None:
270 """Add a flash message to current context."""
271 ctx = AdminContextManager.get_context()
272 if ctx:
273 ctx.add_flash(message, category)
276def get_flashes() -> list[dict[str, str]]:
277 """Get all flash messages from current context."""
278 ctx = AdminContextManager.get_context()
279 if ctx:
280 return ctx.flash_messages
281 return []
284# Breadcrumb helpers
285def breadcrumb(label: str, url: str | None = None) -> None:
286 """Add a breadcrumb to current context."""
287 ctx = AdminContextManager.get_context()
288 if ctx:
289 ctx.add_breadcrumb(label, url)
292def get_breadcrumbs() -> list[dict[str, str]]:
293 """Get breadcrumbs from current context."""
294 ctx = AdminContextManager.get_context()
295 if ctx:
296 return ctx.breadcrumbs
297 return []