Coverage for src / lexigram / admin / controllers / base.py: 38%
123 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-13 22:14 +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 — 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 — non-fatal
166 pass
168 async def render_admin(
169 self,
170 request: Request,
171 content: Any,
172 title: str = "Admin",
173 breadcrumbs: list[dict[str, Any]] | None = None,
174 **extra_context: Any,
175 ) -> HTMLResponse:
176 """Render content within admin shell (async).
178 Args:
179 request: The current request
180 content: Content to render (Component or HTML string)
181 title: Page title
182 breadcrumbs: List of breadcrumb dicts
183 **extra_context: Additional context passed to the renderer.
185 Returns:
186 HTMLResponse with rendered admin page
187 """
188 # Inject runtime theme overrides (primary_color, site_name)
189 await self._apply_theme_overrides(request, extra_context)
191 # If content is awaitable, resolve it first
192 if inspect.isawaitable(content):
193 content = await content
195 # Check for HTMX request targeting #main-content
196 is_htmx = request.headers.get("HX-Request") == "true"
197 target = request.headers.get("HX-Target")
199 if is_htmx and target == "main-content":
200 # Only return the partial content
201 return self.renderer.render_partial(content)
203 return self.renderer.render_page(
204 content,
205 request=request,
206 title=title,
207 breadcrumbs=breadcrumbs,
208 **extra_context,
209 )
211 def flash(self, message: str, category: str = "info") -> None:
212 """Add a flash message to be displayed on next page.
214 Args:
215 message: Message text
216 category: Message category (info, success, warning, error)
217 """
218 self._flash_messages.append({"message": message, "category": category})
220 def get_flash_messages(self) -> list[dict[str, str]]:
221 """Get and clear flash messages.
223 Returns:
224 List of flash message dicts
225 """
226 messages = self._flash_messages.copy()
227 self._flash_messages.clear()
228 return messages
230 def generate_breadcrumbs(
231 self,
232 *crumbs: tuple[str, str],
233 current: str | None = None,
234 ) -> list[dict[str, str]]:
235 """Generate breadcrumb navigation.
237 Args:
238 *crumbs: Variable number of (label, url) tuples
239 current: Label for current page (no link)
241 Returns:
242 List of breadcrumb dicts with 'label' and 'url' keys
244 Example:
245 ```python
246 breadcrumbs = self.generate_breadcrumbs(
247 ("Home", "/admin/"),
248 ("Users", "/admin/users"),
249 current="Edit User"
250 )
251 ```
252 """
253 result = []
255 for label, url in crumbs:
256 result.append({"label": label, "url": url})
258 if current:
259 result.append({"label": current, "url": ""})
261 return result
263 def build_specification(self, request: Request, allowed_fields: list[str]) -> Any:
264 """Build a specification from request query parameters.
266 Args:
267 request: The request
268 allowed_fields: List of fields allowed to be filtered
270 Returns:
271 SpecificationProtocol or None
272 """
273 from lexigram.admin.lib.specifications import (
274 AndSpecification,
275 FieldSpecification,
276 )
278 specs = []
279 for field in allowed_fields:
280 if value := request.query_params.get(field):
281 specs.append(FieldSpecification(field, value)) # type: ignore[abstract]
283 if not specs:
284 return None
286 if len(specs) == 1:
287 return specs[0]
289 return AndSpecification(*specs)
291 async def parallel_fetch(
292 self,
293 *callables: Callable[[], Awaitable[Any]],
294 ) -> list[Any]:
295 """Fetch multiple async operations in parallel.
297 Args:
298 *callables: Async functions to execute concurrently
300 Returns:
301 List of results in same order as input
303 Example:
304 ```python
305 users, posts, comments = await self.parallel_fetch(
306 lambda: user_service.list(),
307 lambda: post_service.list(),
308 lambda: comment_service.list(),
309 )
310 ```
311 """
312 results = await Parallel.gather(*(fn() for fn in callables))
313 return list(results)
315 async def background_task(
316 self,
317 task: Callable[[], Awaitable[Any]],
318 name: str | None = None,
319 ) -> None:
320 """Schedule task to run in background without blocking response.
322 Args:
323 task: Async function to run in background
324 name: Optional task name for tracking
326 Example:
327 ```python
328 # Send email in background
329 await self.background_task(
330 lambda: email_service.send(user, "Welcome!"),
331 name="welcome_email"
332 )
333 # Response returns immediately
334 ```
335 """
336 # Use central TaskManager for background tasks
337 self.task_manager.create_background_task(task(), name=name) # type: ignore[union-attr]
339 def get_routes(self) -> list[Any]:
340 """Extract decorated routes from this controller instance."""
341 from starlette.routing import Route
343 routes = []
345 # In lexigram-web, decorated methods have _route_config
346 for name, method in inspect.getmembers(self, predicate=inspect.ismethod):
347 if hasattr(method, "_route_config"):
348 config = method._route_config
349 # We use the Router._create_endpoint logic to wrap the handler
350 # but since we already have an instance, we can simplify/adapt
352 # Mock a container or just wrap the method directly?
353 # The Router normally wants a class and method name to resolve from container.
354 # But here we already have the instance.
356 # Let's create a compatible Starlette handler
357 async def starlette_handler(request: Request, m=method) -> Any:
358 # We need to handle parameters like Router does
359 # For simplicity, we can reuse Router._create_endpoint logic
360 # Or just call the method if signature allows
361 sig = inspect.signature(m)
362 if "request" in sig.parameters:
363 return await m(request=request)
364 return await m()
366 # Prepend controller prefix to the route path
367 base_path = getattr(self, "prefix", "").rstrip("/")
368 route_path = config["path"]
369 if not route_path.startswith("/"):
370 route_path = f"/{route_path}"
372 if route_path == "/" and base_path:
373 full_path = base_path
374 else:
375 full_path = f"{base_path}{route_path}"
377 if not full_path:
378 full_path = "/"
380 routes.append(
381 Route(
382 full_path,
383 endpoint=starlette_handler,
384 methods=[config["method"]],
385 name=config.get("name") or f"admin_custom_{name}",
386 ),
387 )
389 # Also check for 'index' method if no explicit route matches prefix
390 if hasattr(self, "index") and not any(r.path == "/" for r in routes):
392 async def index_handler(request: Request) -> Any:
393 return await self.index(request)
395 routes.append(
396 Route(
397 "/",
398 endpoint=index_handler,
399 methods=["GET"],
400 name="admin_custom_index",
401 ),
402 )
404 return routes