Coverage for src/lexigram/admin/resources/base.py: 24%
226 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-24 23:31 +0800
1"""Base resource class for Admin Resources.
3.. stability:: stable
5Resources define the configuration for admin UI views including
6columns, actions, filters, and permissions.
7"""
9from __future__ import annotations
11import contextlib
12import re
13from typing import TYPE_CHECKING, Any
14import warnings
16from lexigram.admin.data.data_source import IDataSource
17from lexigram.admin.resources.archive_ops import ArchiveOperationsMixin
18from lexigram.admin.resources.layouts import apply_layout_config
19from lexigram.admin.resources.config import TableConfiguration
21if TYPE_CHECKING:
22 from lexigram.admin.actions.base import HeaderAction
23 from lexigram.admin.forms.components import FormBase
24 from lexigram.admin.layout.layout_manager import LayoutManager
25 from lexigram.admin.rbac.schema import ResourcePermissions
26 from lexigram.admin.relations.manager_ext import RelationManager
27 from lexigram.admin.ui.filters.base import Filter
28 from lexigram.domain import DomainModel
29 from lexigram.ui.actions import Action, BulkAction
30 from lexigram.ui.columns import Column
32_VALID_NAME_RE = re.compile(r"^[a-z][a-z0-9_]*(\.[a-z][a-z0-9_]*)?$")
35def _validate_resource_name(name: str) -> None:
36 """Validate a Resource name is a dotted slug.
38 Raises ``ValueError`` if the name doesn't match
39 the slug pattern (lowercase alphanumeric with underscores,
40 optionally dotted for namespaced resources).
41 """
42 if not _VALID_NAME_RE.match(name):
43 raise ValueError(
44 f"Resource name {name!r} is not a valid slug. "
45 f"Allowed: lowercase alphanumeric + underscores, "
46 f"optionally dotted for namespaced resources. "
47 f"Pattern: {_VALID_NAME_RE.pattern!r}"
48 )
51class Resource(ArchiveOperationsMixin):
52 """Base class for Admin Resources.
54 Resources define the configuration for list views (tables) and form views
55 in the admin interface. Subclass this to create custom resources.
57 Example:
58 >>> class UserResource(Resource):
59 ... model = UserModel
60 ... icon = "users"
61 ... columns = [
62 ... TextColumn("name").sortable(),
63 ... TextColumn("email").sortable(),
64 ... ]
65 ... actions = [EditAction(), DeleteAction()]
66 """
68 # Data Model
69 model: type[DomainModel] | None = None
71 # Registration metadata
72 name: str | None = None
73 cluster: str | None = None
74 """Cluster name for navigation grouping. Replaces ``group``."""
76 # Backward-compat alias for cluster
77 group: str | None = None
78 """Deprecated: use ``cluster`` instead. Kept in sync via __init_subclass__."""
80 # Permissions
81 permissions: ResourcePermissions | None = None
83 # UI Configuration
84 icon: str = "box"
85 label: str | None = None
86 visible_in_sidebar: bool = True
88 # Table Configuration
89 columns: list[Column] = []
90 actions: list[Action] = []
91 action_layout: str = "horizontal"
92 header_actions: list[HeaderAction] = []
93 bulk_actions: list[BulkAction] = []
94 filters: list[Filter] = []
96 # New declarative field system — SchemaField instances
97 # When set, columns and filters are derived automatically.
98 fields: list[Any] = []
100 page_size: int = 20
101 default_sort: str | None = None
103 # Optional default grouping column (users can override via the toolbar)
104 group_by: str | None = None
106 # Table empty-state copy overrides (None = framework defaults)
107 empty_state_title: str | None = None
108 empty_state_message: str | None = None
109 empty_state_icon: str | None = None
111 # Form Configuration
112 form_class: type[FormBase] | None = None
113 # Form display mode: "page" (full page), "modal" (centered modal), "slider" (side panel)
114 form_display_mode: str = "modal" # Options: "page", "modal", "slider"
115 # Model fields excluded from generated forms (e.g. secrets, framework-managed
116 # columns). Resources may extend the default to exclude their own fields.
117 form_exclude_fields: tuple[str, ...] = ("id", "created_at", "updated_at")
119 # Resource Config (Optional fluent config)
120 config: Any = None
122 def __init_subclass__(cls, **kwargs: Any) -> None:
123 """Validate and auto-derive backward-compat attributes when using ``fields``."""
124 super().__init_subclass__(**kwargs)
126 own = cls.__dict__
128 # Sync group <-> cluster for backward compatibility
129 if "group" in own and "cluster" not in own:
130 cls.cluster = own["group"]
131 if "cluster" in own and "group" not in own:
132 cls.group = own["cluster"]
134 # Validate name if explicitly set — must be a dotted slug
135 if "name" in own and own["name"] is not None:
136 _validate_resource_name(own["name"])
138 has_fields = "fields" in own
139 has_columns = "columns" in own
140 has_filters = "filters" in own
141 has_form_class = "form_class" in own
143 if has_fields:
144 if has_columns or has_filters or has_form_class:
145 warnings.warn(
146 "Resource.fields is the new declarative path for schema "
147 "configuration. When fields is set, columns, filters, and "
148 "form_class should not be set — they will be derived from "
149 "fields automatically.",
150 DeprecationWarning,
151 stacklevel=2,
152 )
154 # Derive columns from fields for backward compatibility
155 if not has_columns:
156 cls.columns = list(cls.fields)
158 # Derive filters from fields for backward compatibility
159 if not has_filters:
160 cls.filters = [f for f in cls.fields if getattr(f, "filterable", False)]
162 # Relation managers for inline related-record editing on the ViewPage
163 relations: list[type[RelationManager]] = []
165 # Search Configuration
166 # Fields to include in global search queries. Empty list disables search for this resource.
167 search_fields: list[str] = []
168 # Field used as the display title in search results (falls back to "id")
169 search_title_field: str = "name"
171 # Optional integration knobs
172 cacheable: bool | Any = False # True or CacheableSpec enables list caching
173 searchable: bool | Any = False # True or SearchableSpec enables search index
174 resilient: bool | Any = False # True or ResilientSpec enables retry/circuit
176 def cache_spec(self) -> Any:
177 """Return a CacheableSpec or None based on the cacheable field."""
178 if self.cacheable is False:
179 return None
180 if self.cacheable is True:
181 from lexigram.admin.integrations.cache import CacheableSpec
183 return CacheableSpec()
184 return self.cacheable
186 def search_spec(self) -> Any:
187 """Return a SearchableSpec or None based on the searchable field."""
188 if self.searchable is False:
189 return None
190 if self.searchable is True:
191 from lexigram.contracts.search import SearchableSpec
193 return SearchableSpec(
194 index_name=self.name,
195 fields=tuple(self.search_fields),
196 )
197 return self.searchable
199 def resilient_spec(self) -> Any:
200 """Return a ResilientSpec or None based on the resilient field."""
201 if self.resilient is False:
202 return None
203 if self.resilient is True:
204 from lexigram.admin.integrations.resilience import ResilientSpec
206 return ResilientSpec()
207 return self.resilient
209 # Data source instance for search (set at runtime via set_data_source)
210 _data_source: IDataSource | None = None
212 def set_data_source(self, data_source: IDataSource) -> None:
213 """Attach a data source to this resource for search and list support.
215 Args:
216 data_source: An IDataSource-compatible instance.
218 Raises:
219 TypeError: If data_source does not satisfy IDataSource protocol.
220 """
221 if not isinstance(data_source, IDataSource):
222 raise TypeError(
223 f"data_source must implement IDataSource, got {type(data_source).__name__}"
224 )
225 self._data_source = data_source
227 async def search(self, query: str, *, limit: int = 5) -> list[dict[str, Any]]:
228 """Search this resource for items matching *query*.
230 Override in subclasses for custom search logic. The default
231 implementation queries the attached data source using
232 :attr:`search_fields`.
234 Args:
235 query: Search term entered by the user.
236 limit: Maximum number of results to return.
238 Returns:
239 List of dicts with ``id``, ``title``, and ``subtitle`` keys.
240 """
241 if not self.search_fields or self._data_source is None:
242 return []
244 from lexigram.admin.data.query import QuerySpec
246 qs = (
247 QuerySpec()
248 .with_search(query, self.search_fields)
249 .with_page(1)
250 .with_per_page(limit)
251 )
252 try:
253 result = await self._data_source.find_many(qs)
254 except (AttributeError, TypeError, ValueError, KeyError, RuntimeError):
255 return []
257 hits: list[dict[str, Any]] = []
258 for item in result.items:
259 if isinstance(item, dict):
260 item_id = item.get("id", "")
261 title = (
262 item.get(self.search_title_field)
263 or item.get("name")
264 or item.get("title")
265 or str(item_id)
266 )
267 subtitle = item.get("email") or item.get("description") or ""
268 else:
269 item_id = getattr(item, "id", "")
270 title = (
271 getattr(item, self.search_title_field, None)
272 or getattr(item, "name", None)
273 or str(item_id)
274 )
275 subtitle = (
276 getattr(item, "email", "") or getattr(item, "description", "") or ""
277 )
278 hits.append(
279 {"id": str(item_id), "title": str(title), "subtitle": str(subtitle)}
280 )
281 return hits
283 async def fetch_list(
284 self,
285 *,
286 limit: int = 20,
287 offset: int = 0,
288 filters: dict[str, Any] | None = None,
289 search: str | None = None,
290 search_fields: list[str] | None = None,
291 sort_by: str | None = None,
292 sort_order: str = "asc",
293 include_deleted: bool = False,
294 ) -> tuple[list[Any], int]:
295 """Fetch a paginated list of items via the attached IDataSource.
297 Builds a Query object from the pagination/search/filter/sort parameters
298 and delegates to ``self._data_source.find_many(query)``.
300 Override this in resource subclasses for custom data access logic.
302 Returns:
303 Tuple of (items, total_count).
304 """
305 if self._data_source is None:
306 return [], 0
308 from lexigram.admin.data.query import QuerySpec
310 page = (offset // limit) + 1 if limit else 1
311 qs = QuerySpec().with_page(page).with_per_page(limit)
313 if search and search_fields:
314 qs = qs.with_search(search, search_fields)
316 if sort_by:
317 qs = qs.with_order_by(sort_by, sort_order)
319 if include_deleted:
320 qs = qs.with_deleted(True)
322 for field, value in (filters or {}).items():
323 if isinstance(value, list):
324 qs = qs.with_where_in(field, value)
325 else:
326 qs = qs.with_where_eq(field, value)
328 result = await self._data_source.find_many(qs)
329 items = list(result.items)
330 total = result.total
331 return items, total
333 async def before_create(self, data: dict) -> dict:
334 """Hook called before a record is created.
336 Args:
337 data: Record data to be created
339 Returns:
340 Modified data
341 """
342 return data
344 async def before_validate(self, data: dict) -> Any:
345 """Validate and coerce form data against the resource model.
347 Base implementation coerces HTML form strings to proper Python types
348 via _coerce_form_data, then validates against ``self.model``.
349 Returns Ok(coerced_data) on success, Err(AdminValidationError) with
350 per-field errors on failure.
352 Override in subclasses to add custom validation logic.
353 """
354 from lexigram.admin.exceptions import AdminValidationError
355 from lexigram.admin.resources.form_coercion import _coerce_form_data
356 from lexigram.contracts.exceptions.domain import FieldError
357 from lexigram.result import Err, Ok
359 coerced = _coerce_form_data(data, self.model)
360 if self.model is None:
361 return Ok(coerced)
363 if not hasattr(self.model, "model_validate"):
364 return Ok(coerced)
366 try:
367 self.model.model_validate(coerced)
368 except (ValueError, TypeError) as exc:
369 msg = str(exc)
370 errors: list[FieldError] = []
372 is_pydantic = (
373 type(exc).__name__ == "ValidationError"
374 and "pydantic" in type(exc).__module__
375 )
376 if is_pydantic:
377 for err in exc.errors(): # type: ignore[union-attr]
378 field = str(err["loc"][0]) if err.get("loc") else None
379 if field and field in coerced:
380 errors.append(FieldError(field=field, message=err["msg"]))
381 else:
382 field = None
383 if msg.startswith("Field '"):
384 field = msg.split("'")[1]
385 if field:
386 errors.append(FieldError(field=field, message=msg))
388 if errors:
389 return Err(
390 AdminValidationError(
391 message="Form validation failed",
392 errors=errors,
393 )
394 )
395 return Ok(coerced)
397 return Ok(coerced)
399 async def after_create(self, record: Any) -> None:
400 """Hook called after a record is created.
402 Args:
403 record: Created record
404 """
406 async def before_update(self, item_id: Any, data: dict) -> dict:
407 """Hook called before a record is updated.
409 Args:
410 item_id: Record identifier
411 data: Updated record data
413 Returns:
414 Modified data
415 """
416 return data
418 async def after_update(self, record: Any) -> None:
419 """Hook called after a record is updated.
421 Args:
422 record: Updated record
423 """
425 async def before_delete(self, item_id: Any) -> None:
426 """Hook called before a record is deleted.
428 Args:
429 item_id: Record identifier
430 """
432 async def after_delete(self, item_id: Any) -> None:
433 """Hook called after a record is deleted.
435 Args:
436 item_id: Record identifier
437 """
439 @classmethod
440 def get_action_hooks(cls, action_name: str) -> list[Any]:
441 """Get action lifecycle hooks for the named action.
443 Override in a resource subclass to attach ``ActionHookProtocol``
444 hooks to registry-based actions. Hooks are collected by
445 ``ActionExecutor`` and run before/after the action body and on
446 failure.
448 Args:
449 action_name: Name of the action (e.g. ``"export"``)
451 Returns:
452 List of action hooks for the action.
453 """
454 return []
456 @classmethod
457 def get_table_config(cls) -> TableConfiguration:
458 """Get the table configuration for this resource.
460 Returns:
461 TableConfiguration with columns, actions, filters
462 """
463 cfg = cls.config
465 # Resolve configuration with priority: Config Object > Class Attribute
466 per_page = (
467 cls._get_config_value(cfg, "per_page", cls.page_size)
468 if cfg
469 else cls.page_size
470 )
471 default_sort = (
472 cls._get_config_value(
473 cfg,
474 "default_sort_field",
475 cls.default_sort,
476 )
477 if cfg
478 else cls.default_sort
479 )
480 default_sort_order = (
481 cls._get_config_value(cfg, "default_sort_order", "asc") if cfg else "asc"
482 )
483 action_layout = (
484 cls._get_config_value(cfg, "action_layout", cls.action_layout)
485 if cfg
486 else cls.action_layout
487 )
489 resource_name = cls.label or cls.__name__.replace("Resource", "")
490 if cfg and cfg.display_name:
491 resource_name = cfg.display_name
493 columns = list(
494 cls._get_config_value(cfg, "columns", cls.columns) if cfg else cls.columns
495 )
496 actions = list(
497 cls._get_config_value(cfg, "actions", cls.actions) if cfg else cls.actions
498 )
499 filters = list(
500 cls._get_config_value(cfg, "filters_list", cls.filters)
501 if cfg
502 else cls.filters
503 )
505 # Check class attribute `layout_type` as fallback for legacy resources
506 layout_fallback = getattr(cls, "layout_type", "stack")
507 default_layout = (
508 cls._get_config_value(cfg, "layout", layout_fallback)
509 if cfg
510 else layout_fallback
511 )
512 # Check class attribute `data_view` as fallback for legacy resources
513 view_fallback = getattr(cls, "data_view", "tabular")
514 default_view = (
515 cls._get_config_value(cfg, "view", view_fallback) if cfg else view_fallback
516 )
518 empty_state_title = (
519 cls._get_config_value(cfg, "empty_state_title", cls.empty_state_title)
520 if cfg
521 else cls.empty_state_title
522 )
523 empty_state_message = (
524 cls._get_config_value(cfg, "empty_state_message", cls.empty_state_message)
525 if cfg
526 else cls.empty_state_message
527 )
528 empty_state_icon = (
529 cls._get_config_value(cfg, "empty_state_icon", cls.empty_state_icon)
530 if cfg
531 else cls.empty_state_icon
532 )
533 group_by = (
534 cls._get_config_value(cfg, "group_by", cls.group_by)
535 if cfg
536 else cls.group_by
537 )
539 return TableConfiguration(
540 columns=columns,
541 actions=actions,
542 header_actions=list(cls.header_actions),
543 bulk_actions=list(cls.bulk_actions),
544 filter_options=filters,
545 per_page=per_page,
546 default_sort_by=default_sort,
547 default_sort_order=default_sort_order,
548 resource_name=resource_name,
549 action_layout=action_layout,
550 default_layout=default_layout,
551 default_view=default_view,
552 empty_state_title=empty_state_title,
553 empty_state_message=empty_state_message,
554 empty_state_icon=empty_state_icon,
555 group_by=group_by,
556 )
558 @classmethod
559 def get_form_class(cls) -> type[FormBase] | None:
560 """Return the Form class to use for create/edit views.
562 Returns:
563 Form class or None
564 """
565 return cls.form_class
567 @classmethod
568 def get_form_display_mode(cls) -> str:
569 """Return the form display mode for create/edit views.
571 Returns:
572 Display mode: "page", "modal", or "slider"
573 """
574 cfg = cls.config
575 return (
576 cls._get_config_value(cfg, "form_display_mode", cls.form_display_mode)
577 if cfg
578 else cls.form_display_mode
579 )
581 @classmethod
582 def get_layout_manager(cls) -> LayoutManager:
583 """Get layout manager with configured views.
585 Returns:
586 LayoutManager instance
587 """
588 from lexigram.admin.layout import LayoutManager
590 manager = LayoutManager()
591 cfg = cls.config
593 if cfg and cfg.views_list:
594 for view in cfg.views_list:
595 if hasattr(view, "to_config"):
596 layout_config = view.to_config()
597 apply_layout_config(manager, layout_config)
599 # Set default view
600 if cfg.view:
601 with contextlib.suppress(ValueError):
602 manager.set_default(cfg.view)
604 return manager
606 @staticmethod
607 def _get_config_value(cfg: Any, attr: str, default: Any) -> Any:
608 """Get configuration value with fallback to default or private attribute.
610 Args:
611 cfg: Configuration object
612 attr: Attribute name
613 default: Default value if not found
615 Returns:
616 Configuration value or default
617 """
618 if cfg is None:
619 return default
621 # Try public attribute/property
622 val = getattr(cfg, attr, None)
624 # If it's the fluent method (callable) or missing, try the private attribute
625 if val is None or callable(val):
626 val = getattr(cfg, f"_{attr}", None)
628 return val if val is not None else default
631__all__ = ["Resource"]