Coverage for src / lexigram / admin / state / context.py: 44%

1"""Request context for lexigram-admin. 

2 

3Provides a context manager for request-scoped state that integrates 

4with lexigram-cache RequestContext for caching and with HTMX patterns. 

5""" 

6 

7from __future__ import annotations 

8 

9from contextvars import ContextVar 

10from dataclasses import dataclass, field 

11from functools import wraps 

12from typing import TYPE_CHECKING, Any, ClassVar, TypeVar 

13 

14from lexigram.logging import get_logger 

15 

16if TYPE_CHECKING: 

17 from collections.abc import Awaitable, Callable 

18 

19 from starlette.requests import Request 

20 

21logger = get_logger(__name__) 

22 

23T = TypeVar("T") 

24 

25 

26@dataclass 

27class AdminContext: 

28 """Request-scoped context for admin operations. 

29 

30 Holds all request-related state in a single immutable container. 

31 This avoids mutation issues and makes state flow explicit. 

32 

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 """ 

46 

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) 

58 

59 @property 

60 def is_authenticated(self) -> bool: 

61 """Check if user is authenticated.""" 

62 return self.user is not None 

63 

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) 

70 

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) 

76 

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}) 

80 

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] 

84 

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 ) 

105 

106 

107@dataclass 

108class HTMXInfo: 

109 """Information about HTMX request.""" 

110 

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 

118 

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 ) 

135 

136 

137def wants_fragment(request: Request) -> bool: 

138 """Return True when the request expects a fragment swap. 

139 

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") 

146 

147 

148class AdminContextManager: 

149 """Context manager for admin request context. 

150 

151 Usage: 

152 async with AdminContextManager(request) as ctx: 

153 # ctx is AdminContext with user, permissions, etc. 

154 pass 

155 """ 

156 

157 _context_var: ClassVar[ContextVar[AdminContext | None]] = ContextVar( 

158 "admin_context", 

159 default=None, 

160 ) 

161 

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() 

166 

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 

176 

177 def __init__(self, request: Request): 

178 self.request = request 

179 self.ctx: AdminContext | None = None 

180 self._token: Any = None 

181 

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 [] 

191 

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: 

203 pass 

204 

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) 

210 

211 # Create HTMX info 

212 htmx = HTMXInfo.from_request(self.request) 

213 

214 # Create cache context if available 

215 cache_ctx = None 

216 try: 

217 from lexigram.primitives.context import RequestContext 

218 

219 cache_ctx = RequestContext() # type: ignore[call-arg] 

220 except (RuntimeError, ImportError, TypeError): 

221 pass 

222 

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 ) 

231 

232 # Restore flash messages from session 

233 self.ctx.flash_messages = self._read_flash_from_session() 

234 

235 # Set in context var 

236 self._token = AdminContextManager._context_var.set(self.ctx) 

237 

238 return self.ctx 

239 

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 

248 

249 

250def with_context(func: Callable[..., Awaitable[T]]) -> Callable[..., Awaitable[T]]: 

251 """Decorator to automatically set up admin context. 

252 

253 Usage: 

254 @with_context 

255 async def my_handler(request: Request, ctx: AdminContext): 

256 # ctx is automatically provided 

257 pass 

258 """ 

259 

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) 

264 

265 return wrapper 

266 

267 

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) 

274 

275 

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 [] 

282 

283 

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) 

290 

291 

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 []