Coverage for src/lexigram/admin/controllers/base.py: 17%
176 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-24 23:39 +0800
1"""Base controller classes for Lexigram Admin.
3This module provides base classes that leverage Lexigram's DI container
4to provide common functionality to all admin controllers.
5"""
7from __future__ import annotations
9from collections.abc import Awaitable, Callable
10import inspect
11from typing import Any
13from starlette.requests import Request
14from starlette.responses import HTMLResponse
16from lexigram.admin.auth.models import AdminUser
17from lexigram.admin.engine.renderer import AdminRenderer
18from lexigram.admin.middleware.auth import current_user
19from lexigram.concurrency import Parallel
20from lexigram.contracts.core import TaskManagerProtocol
21from lexigram.contracts.web.controller import ControllerProtocol
22from lexigram.di.decorators import inject
25@inject
26class AdminController(ControllerProtocol):
27 """Base async controller for admin pages with authentication and rendering helpers.
29 This base class provides:
30 - Access to AdminRenderer for rendering pages
31 - current_user() helper for auth
32 - render_admin() for consistent page rendering (async)
33 - Flash message support
34 - Breadcrumb generation
35 - Parallel async operations
36 - Background task scheduling
38 All admin controllers should inherit from this to get these features.
40 Example:
41 ```python
42 class MyAdminController(AdminController):
43 def __init__(self, renderer: AdminRenderer):
44 super().__init__(renderer)
46 @get("/admin/dashboard")
47 async def dashboard(self, request: Request):
48 user = self.current_user(request)
49 content = f"Welcome {user.name}!"
50 return await self.render_admin(request, content)
51 ```
52 """
54 def __init__(
55 self,
56 renderer: AdminRenderer,
57 task_manager: TaskManagerProtocol | None = None,
58 settings_service: Any | None = None,
59 ):
60 """Initialize admin controller.
62 Args:
63 renderer: AdminRenderer instance (DI-injected)
64 task_manager: TaskManagerProtocol instance (optional)
65 settings_service: AdminSettingsService instance (optional), for
66 runtime theme overrides (site_name, primary_color).
67 """
68 self.renderer = renderer
69 self.task_manager = task_manager
70 self._settings_service = settings_service
71 self._flash_messages: list[dict[str, str]] = []
73 @classmethod
74 def collect_routes(cls) -> list[dict[str, Any]]:
75 """Collect routes from controller methods."""
76 routes = []
77 seen_handlers = set()
79 for klass in cls.__mro__:
80 if klass is object:
81 continue
83 for attr_name in dir(klass):
84 if attr_name.startswith("_") or attr_name in seen_handlers:
85 continue
87 attr_value = getattr(klass, attr_name, None)
88 if attr_value is not None and hasattr(attr_value, "_route_config"):
89 route_config = attr_value._route_config
90 routes.append(
91 {
92 "method": route_config["method"],
93 "path": route_config["path"],
94 "handler_name": attr_name,
95 "response_model": route_config.get("response_model"),
96 "request_model": route_config.get("request_model"),
97 "status_code": route_config.get("status_code", 200),
98 "summary": route_config.get("summary"),
99 "description": route_config.get("description"),
100 "tags": route_config.get("tags"),
101 "operation_id": route_config.get("operation_id"),
102 "responses": route_config.get("responses"),
103 "deprecated": route_config.get("deprecated", False),
104 }
105 )
106 seen_handlers.add(attr_name)
108 return routes
110 def current_user(self, request: Request) -> AdminUser:
111 """Get the current authenticated user.
113 Args:
114 request: The current request
116 Returns:
117 AdminUser instance or GUEST_USER if not authenticated
118 """
119 return current_user(request) # type: ignore[return-value]
121 async def _apply_theme_overrides(
122 self,
123 request: Request,
124 extra_context: dict[str, Any],
125 ) -> None:
126 """Load runtime theme settings and merge into extra_context.
128 Uses the controller's settings service when injected, otherwise
129 builds one from the request-scoped DI container (mirroring the
130 bundle's own construction) so every renderer path honors the same
131 persisted branding.
132 """
133 if not self._settings_service:
134 try:
135 from lexigram.admin.services.settings_service import (
136 resolve_admin_settings_service,
137 )
139 container = getattr(request.state, "container", None) or getattr(
140 request.app.state, "container", None
141 )
142 if container is not None:
143 self._settings_service = await resolve_admin_settings_service(
144 container
145 )
146 except Exception: # noqa: BLE001, S110 — non-fatal
147 pass
148 if not self._settings_service:
149 return
150 try:
151 from lexigram.admin.multitenancy.adapter import resolve_tenant_id
153 tenant = await resolve_tenant_id(request, default="default")
154 overrides = await self._settings_service.get_all(tenant)
155 for field in (
156 "primary_color",
157 "site_name",
158 "logo_url",
159 "favicon_url",
160 "dark_mode",
161 ):
162 value = overrides.get(field) or overrides.get(f"admin.branding.{field}")
163 if value:
164 extra_context.setdefault(field, value)
165 except Exception: # noqa: BLE001, S110 — non-fatal
166 pass
168 async def _apply_tenant_context(
169 self,
170 request: Request,
171 extra_context: dict[str, Any],
172 ) -> None:
173 """Resolve current tenant + switchable list into extra_context.
175 Populates ``current_tenant_id``/``current_tenant_name``/
176 ``tenant_list``/``tenant_csrf_token`` only when tenancy is enabled
177 and the requesting user is a superadmin — presence of
178 ``current_tenant_id`` doubles as the render gate for
179 ``TenantSwitcher`` (absent means "don't show the switcher"), so no
180 separate flag is threaded through.
182 ``tenant_csrf_token`` is generated fresh via the CSRF service
183 rather than reusing ``request.state.csrf_token``, because that
184 value is only populated by resource CRUD form rendering
185 (``resources/form_renderer.py``, ``resources/handler.py``), not on
186 the dashboard/widget pages this switcher actually appears on.
187 """
188 try:
189 from lexigram.admin.config import AdminConfig
191 container = getattr(request.state, "container", None) or getattr(
192 request.app.state, "container", None
193 )
194 if container is None:
195 return
196 config = await container.resolve(AdminConfig)
197 if not config.tenancy.enabled:
198 return
200 user = getattr(request.state, "user", None)
201 if not user:
202 return
204 from lexigram.admin.rbac.super_admin import is_super_admin
206 if not is_super_admin(user, config.rbac.super_admin_role):
207 return
209 from lexigram.admin.multitenancy.adapter import (
210 TenantProviderRegistry,
211 resolve_tenant_id,
212 )
214 registry = await container.resolve(TenantProviderRegistry)
215 current_tenant_id = await resolve_tenant_id(request, default="default")
216 current_tenant = await registry.get(current_tenant_id)
217 extra_context.setdefault("current_tenant_id", current_tenant_id)
218 extra_context.setdefault(
219 "current_tenant_name",
220 current_tenant.name if current_tenant else current_tenant_id,
221 )
222 extra_context.setdefault(
223 "tenant_list",
224 [(t.tenant_id, t.name) for t in await registry.all(active_only=True)],
225 )
227 try:
228 from lexigram.admin.auth.protocols import AdminCsrfServiceProtocol
230 csrf_service = await container.resolve(AdminCsrfServiceProtocol)
231 session = getattr(request, "session", {})
232 session_id = session.get("admin_user_id", "")
233 extra_context.setdefault(
234 "tenant_csrf_token", csrf_service.generate_token(session_id)
235 )
236 except Exception: # noqa: BLE001, S110 — token generation is non-fatal
237 pass
238 except Exception: # noqa: BLE001, S110 — matches _apply_theme_overrides
239 pass
241 async def _apply_impersonation_context(
242 self,
243 request: Request,
244 extra_context: dict[str, Any],
245 ) -> None:
246 """Populate impersonation banner state into extra_context, if active.
248 Resolves ``ImpersonationService`` from the request-scoped container
249 and checks whether the current actor has an active session. Display
250 name resolution is out of scope (no user-store dependency here) —
251 the banner shows the raw target user ID.
252 """
253 user = getattr(request.state, "user", None)
254 if user is None:
255 return
257 try:
258 from lexigram.admin.services.impersonation import ImpersonationService
260 container = getattr(request.state, "container", None) or getattr(
261 request.app.state, "container", None
262 )
263 if container is None:
264 return
265 service = await container.resolve(ImpersonationService)
266 if service is None:
267 return
269 actor_id = str(getattr(user, "id", ""))
270 session = service.get_active_session(actor_id, request)
271 if session is not None:
272 extra_context.setdefault("impersonation_active", True)
273 extra_context.setdefault(
274 "impersonation_target_id", session.target_user_id
275 )
276 except Exception: # noqa: BLE001, S110 — non-fatal
277 pass
279 async def render_admin(
280 self,
281 request: Request,
282 content: Any,
283 title: str = "Admin",
284 breadcrumbs: list[dict[str, Any]] | None = None,
285 **extra_context: Any,
286 ) -> HTMLResponse:
287 """Render content within admin shell (async).
289 Args:
290 request: The current request
291 content: Content to render (Component or HTML string)
292 title: Page title
293 breadcrumbs: List of breadcrumb dicts
294 **extra_context: Additional context passed to the renderer.
296 Returns:
297 HTMLResponse with rendered admin page
298 """
299 # Inject runtime theme overrides (primary_color, site_name)
300 await self._apply_theme_overrides(request, extra_context)
301 # Inject tenant context (current tenant, switchable list, CSRF token)
302 await self._apply_tenant_context(request, extra_context)
303 # Inject impersonation banner state, if an active session exists
304 await self._apply_impersonation_context(request, extra_context)
306 # If content is awaitable, resolve it first
307 if inspect.isawaitable(content):
308 content = await content
310 # Check for HTMX request targeting #main-content
311 is_htmx = request.headers.get("HX-Request") == "true"
312 target = request.headers.get("HX-Target")
314 if is_htmx and target == "main-content":
315 # Only return the partial content
316 return self.renderer.render_partial(content)
318 return self.renderer.render_page(
319 content,
320 request=request,
321 title=title,
322 breadcrumbs=breadcrumbs,
323 **extra_context,
324 )
326 def flash(self, message: str, category: str = "info") -> None:
327 """Add a flash message to be displayed on next page.
329 Args:
330 message: Message text
331 category: Message category (info, success, warning, error)
332 """
333 self._flash_messages.append({"message": message, "category": category})
335 def get_flash_messages(self) -> list[dict[str, str]]:
336 """Get and clear flash messages.
338 Returns:
339 List of flash message dicts
340 """
341 messages = self._flash_messages.copy()
342 self._flash_messages.clear()
343 return messages
345 def generate_breadcrumbs(
346 self,
347 *crumbs: tuple[str, str],
348 current: str | None = None,
349 ) -> list[dict[str, str]]:
350 """Generate breadcrumb navigation.
352 Args:
353 *crumbs: Variable number of (label, url) tuples
354 current: Label for current page (no link)
356 Returns:
357 List of breadcrumb dicts with 'label' and 'url' keys
359 Example:
360 ```python
361 breadcrumbs = self.generate_breadcrumbs(
362 ("Home", "/admin/"),
363 ("Users", "/admin/users"),
364 current="Edit User"
365 )
366 ```
367 """
368 result = []
370 for label, url in crumbs:
371 result.append({"label": label, "url": url})
373 if current:
374 result.append({"label": current, "url": ""})
376 return result
378 def build_specification(self, request: Request, allowed_fields: list[str]) -> Any:
379 """Build a specification from request query parameters.
381 Args:
382 request: The request
383 allowed_fields: List of fields allowed to be filtered
385 Returns:
386 SpecificationProtocol or None
387 """
388 from lexigram.admin.lib.specifications import (
389 AndSpecification,
390 FieldSpecification,
391 )
393 specs = []
394 for field in allowed_fields:
395 if value := request.query_params.get(field):
396 specs.append(FieldSpecification(field, value)) # type: ignore[abstract]
398 if not specs:
399 return None
401 if len(specs) == 1:
402 return specs[0]
404 return AndSpecification(*specs)
406 async def parallel_fetch(
407 self,
408 *callables: Callable[[], Awaitable[Any]],
409 ) -> list[Any]:
410 """Fetch multiple async operations in parallel.
412 Args:
413 *callables: Async functions to execute concurrently
415 Returns:
416 List of results in same order as input
418 Example:
419 ```python
420 users, posts, comments = await self.parallel_fetch(
421 lambda: user_service.list(),
422 lambda: post_service.list(),
423 lambda: comment_service.list(),
424 )
425 ```
426 """
427 results = await Parallel.gather(*(fn() for fn in callables))
428 return list(results)
430 async def background_task(
431 self,
432 task: Callable[[], Awaitable[Any]],
433 name: str | None = None,
434 ) -> None:
435 """Schedule task to run in background without blocking response.
437 Args:
438 task: Async function to run in background
439 name: Optional task name for tracking
441 Example:
442 ```python
443 # Send email in background
444 await self.background_task(
445 lambda: email_service.send(user, "Welcome!"),
446 name="welcome_email"
447 )
448 # Response returns immediately
449 ```
450 """
451 # Use central TaskManager for background tasks
452 self.task_manager.create_background_task(task(), name=name) # type: ignore[union-attr]
454 def get_routes(self) -> list[Any]:
455 """Extract decorated routes from this controller instance."""
456 from starlette.routing import Route
458 routes = []
460 # In lexigram-web, decorated methods have _route_config
461 for name, method in inspect.getmembers(self, predicate=inspect.ismethod):
462 if hasattr(method, "_route_config"):
463 config = method._route_config
464 # We use the Router._create_endpoint logic to wrap the handler
465 # but since we already have an instance, we can simplify/adapt
467 # Mock a container or just wrap the method directly?
468 # The Router normally wants a class and method name to resolve from container.
469 # But here we already have the instance.
471 # Let's create a compatible Starlette handler
472 async def starlette_handler(request: Request, m=method) -> Any:
473 # We need to handle parameters like Router does
474 # For simplicity, we can reuse Router._create_endpoint logic
475 # Or just call the method if signature allows
476 sig = inspect.signature(m)
477 if "request" in sig.parameters:
478 return await m(request=request)
479 return await m()
481 # Prepend controller prefix to the route path
482 base_path = getattr(self, "prefix", "").rstrip("/")
483 route_path = config["path"]
484 if not route_path.startswith("/"):
485 route_path = f"/{route_path}"
487 if route_path == "/" and base_path:
488 full_path = base_path
489 else:
490 full_path = f"{base_path}{route_path}"
492 if not full_path:
493 full_path = "/"
495 routes.append(
496 Route(
497 full_path,
498 endpoint=starlette_handler,
499 methods=[config["method"]],
500 name=config.get("name") or f"admin_custom_{name}",
501 ),
502 )
504 # Also check for 'index' method if no explicit route matches prefix
505 if hasattr(self, "index") and not any(r.path == "/" for r in routes):
507 async def index_handler(request: Request) -> Any:
508 return await self.index(request)
510 routes.append(
511 Route(
512 "/",
513 endpoint=index_handler,
514 methods=["GET"],
515 name="admin_custom_index",
516 ),
517 )
519 return routes