Coverage for src / lexigram / admin / resources / base.py: 32%
292 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-13 22:14 +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.config import TableConfiguration
19if TYPE_CHECKING:
20 from lexigram.admin.forms.components import FormBase
21 from lexigram.admin.layout.layout_manager import LayoutManager
22 from lexigram.admin.rbac.schema import ResourcePermissions
23 from lexigram.admin.relations.manager_ext import RelationManager
24 from lexigram.admin.ui.filters.base import Filter
25 from lexigram.domain import DomainModel
26 from lexigram.ui.actions import Action, BulkAction
27 from lexigram.ui.columns import Column
29_VALID_NAME_RE = re.compile(r"^[a-z][a-z0-9_]*(\.[a-z][a-z0-9_]*)?$")
32def _validate_resource_name(name: str) -> None:
33 """Validate a Resource name is a dotted slug.
35 Raises ``ValueError`` if the name doesn't match
36 the slug pattern (lowercase alphanumeric with underscores,
37 optionally dotted for namespaced resources).
38 """
39 if not _VALID_NAME_RE.match(name):
40 raise ValueError(
41 f"Resource name {name!r} is not a valid slug. "
42 f"Allowed: lowercase alphanumeric + underscores, "
43 f"optionally dotted for namespaced resources. "
44 f"Pattern: {_VALID_NAME_RE.pattern!r}"
45 )
48class Resource:
49 """Base class for Admin Resources.
51 Resources define the configuration for list views (tables) and form views
52 in the admin interface. Subclass this to create custom resources.
54 Example:
55 >>> class UserResource(Resource):
56 ... model = UserModel
57 ... icon = "users"
58 ... columns = [
59 ... TextColumn("name").sortable(),
60 ... TextColumn("email").sortable(),
61 ... ]
62 ... actions = [EditAction(), DeleteAction()]
63 """
65 # Data Model
66 model: type[DomainModel] | None = None
68 # Registration metadata
69 name: str | None = None
70 cluster: str | None = None
71 """Cluster name for navigation grouping. Replaces ``group``."""
73 # Backward-compat alias for cluster
74 group: str | None = None
75 """Deprecated: use ``cluster`` instead. Kept in sync via __init_subclass__."""
77 # Permissions
78 permissions: ResourcePermissions | None = None
80 # UI Configuration
81 icon: str = "box"
82 label: str | None = None
83 visible_in_sidebar: bool = True
85 # Table Configuration
86 columns: list[Column] = []
87 actions: list[Action] = []
88 action_layout: str = "horizontal"
89 bulk_actions: list[BulkAction] = []
90 filters: list[Filter] = []
92 # New declarative field system — SchemaField instances
93 # When set, columns and filters are derived automatically.
94 fields: list[Any] = []
96 page_size: int = 20
97 default_sort: str | None = None
99 # Form Configuration
100 form_class: type[FormBase] | None = None
101 # Form display mode: "page" (full page), "modal" (centered modal), "slider" (side panel)
102 form_display_mode: str = "modal" # Options: "page", "modal", "slider"
104 # Resource Config (Optional fluent config)
105 config: Any = None
107 def __init_subclass__(cls, **kwargs: Any) -> None:
108 """Validate and auto-derive backward-compat attributes when using ``fields``."""
109 super().__init_subclass__(**kwargs)
111 own = cls.__dict__
113 # Sync group <-> cluster for backward compatibility
114 if "group" in own and "cluster" not in own:
115 cls.cluster = own["group"]
116 if "cluster" in own and "group" not in own:
117 cls.group = own["cluster"]
119 # Validate name if explicitly set — must be a dotted slug
120 if "name" in own and own["name"] is not None:
121 _validate_resource_name(own["name"])
123 has_fields = "fields" in own
124 has_columns = "columns" in own
125 has_filters = "filters" in own
126 has_form_class = "form_class" in own
128 if has_fields:
129 if has_columns or has_filters or has_form_class:
130 warnings.warn(
131 "Resource.fields is the new declarative path for schema "
132 "configuration. When fields is set, columns, filters, and "
133 "form_class should not be set — they will be derived from "
134 "fields automatically.",
135 DeprecationWarning,
136 stacklevel=2,
137 )
139 # Derive columns from fields for backward compatibility
140 if not has_columns:
141 cls.columns = list(cls.fields)
143 # Derive filters from fields for backward compatibility
144 if not has_filters:
145 cls.filters = [f for f in cls.fields if getattr(f, "filterable", False)]
147 # Relation managers for inline related-record editing on the ViewPage
148 relations: list[type[RelationManager]] = []
150 # Search Configuration
151 # Fields to include in global search queries. Empty list disables search for this resource.
152 search_fields: list[str] = []
153 # Field used as the display title in search results (falls back to "id")
154 search_title_field: str = "name"
156 # Optional integration knobs
157 cacheable: bool | Any = False # True or CacheableSpec enables list caching
158 searchable: bool | Any = False # True or SearchableSpec enables search index
159 resilient: bool | Any = False # True or ResilientSpec enables retry/circuit
161 def cache_spec(self) -> Any:
162 """Return a CacheableSpec or None based on the cacheable field."""
163 if self.cacheable is False:
164 return None
165 if self.cacheable is True:
166 from lexigram.admin.integrations.cache import CacheableSpec
168 return CacheableSpec()
169 return self.cacheable
171 def search_spec(self) -> Any:
172 """Return a SearchableSpec or None based on the searchable field."""
173 if self.searchable is False:
174 return None
175 if self.searchable is True:
176 from lexigram.admin.integrations.search import SearchableSpec
178 return SearchableSpec(
179 index_name=self.name,
180 fields=tuple(self.search_fields),
181 )
182 return self.searchable
184 def resilient_spec(self) -> Any:
185 """Return a ResilientSpec or None based on the resilient field."""
186 if self.resilient is False:
187 return None
188 if self.resilient is True:
189 from lexigram.admin.integrations.resilience import ResilientSpec
191 return ResilientSpec()
192 return self.resilient
194 # Data source instance for search (set at runtime via set_data_source)
195 _data_source: IDataSource | None = None
197 def set_data_source(self, data_source: IDataSource) -> None:
198 """Attach a data source to this resource for search and list support.
200 Args:
201 data_source: An IDataSource-compatible instance.
203 Raises:
204 TypeError: If data_source does not satisfy IDataSource protocol.
205 """
206 if not isinstance(data_source, IDataSource):
207 raise TypeError(
208 f"data_source must implement IDataSource, got {type(data_source).__name__}"
209 )
210 self._data_source = data_source
212 async def search(self, query: str, *, limit: int = 5) -> list[dict[str, Any]]:
213 """Search this resource for items matching *query*.
215 Override in subclasses for custom search logic. The default
216 implementation queries the attached data source using
217 :attr:`search_fields`.
219 Args:
220 query: Search term entered by the user.
221 limit: Maximum number of results to return.
223 Returns:
224 List of dicts with ``id``, ``title``, and ``subtitle`` keys.
225 """
226 if not self.search_fields or self._data_source is None:
227 return []
229 from lexigram.admin.data.query import QuerySpec
231 qs = (
232 QuerySpec()
233 .with_search(query, self.search_fields)
234 .with_page(1)
235 .with_per_page(limit)
236 )
237 try:
238 result = await self._data_source.find_many(qs)
239 except (AttributeError, TypeError, ValueError, KeyError, RuntimeError):
240 return []
242 hits: list[dict[str, Any]] = []
243 for item in result.items:
244 if isinstance(item, dict):
245 item_id = item.get("id", "")
246 title = (
247 item.get(self.search_title_field)
248 or item.get("name")
249 or item.get("title")
250 or str(item_id)
251 )
252 subtitle = item.get("email") or item.get("description") or ""
253 else:
254 item_id = getattr(item, "id", "")
255 title = (
256 getattr(item, self.search_title_field, None)
257 or getattr(item, "name", None)
258 or str(item_id)
259 )
260 subtitle = (
261 getattr(item, "email", "") or getattr(item, "description", "") or ""
262 )
263 hits.append(
264 {"id": str(item_id), "title": str(title), "subtitle": str(subtitle)}
265 )
266 return hits
268 async def fetch_list(
269 self,
270 *,
271 limit: int = 20,
272 offset: int = 0,
273 filters: dict[str, Any] | None = None,
274 search: str | None = None,
275 search_fields: list[str] | None = None,
276 sort_by: str | None = None,
277 sort_order: str = "asc",
278 include_deleted: bool = False,
279 ) -> tuple[list[Any], int]:
280 """Fetch a paginated list of items via the attached IDataSource.
282 Builds a Query object from the pagination/search/filter/sort parameters
283 and delegates to ``self._data_source.find_many(query)``.
285 Override this in resource subclasses for custom data access logic.
287 Returns:
288 Tuple of (items, total_count).
289 """
290 if self._data_source is None:
291 return [], 0
293 from lexigram.admin.data.query import QuerySpec
295 page = (offset // limit) + 1 if limit else 1
296 qs = QuerySpec().with_page(page).with_per_page(limit)
298 if search and search_fields:
299 qs = qs.with_search(search, search_fields)
301 if sort_by:
302 qs = qs.with_order_by(sort_by, sort_order)
304 if include_deleted:
305 qs = qs.with_deleted(True)
307 for field, value in (filters or {}).items():
308 if isinstance(value, list):
309 qs = qs.with_where_in(field, value)
310 else:
311 qs = qs.with_where_eq(field, value)
313 result = await self._data_source.find_many(qs)
314 items = list(result.items)
315 total = result.total
316 return items, total
318 async def before_clone(self, data: dict) -> dict:
319 """Hook called before a record is cloned.
321 Strips the ``id`` field (so a new ID is assigned) and
322 appends `` (Copy)`` to the ``name`` field. Override
323 to customise clone behaviour.
325 Args:
326 data: Record data dict fetched from the data source.
328 Returns:
329 Modified data dict to be passed to ``create``.
330 """
331 data.pop("id", None)
332 if "name" in data:
333 data["name"] = f"{data['name']} (Copy)"
334 return data
336 async def after_clone(self, record: Any) -> None:
337 """Hook called after a record has been cloned.
339 Args:
340 record: The newly created record returned by the data source.
341 """
343 async def duplicate(self, item_id: Any) -> Any:
344 """Duplicate (clone) a record by its identifier.
346 Fetches the existing record via the attached data source,
347 calls :meth:`before_clone` to prepare the data, creates
348 a new record, and calls :meth:`after_clone` with the result.
350 Args:
351 item_id: Identifier of the record to clone.
353 Returns:
354 The newly created record.
356 Raises:
357 RuntimeError: If no data source is attached.
358 """
359 if self._data_source is None:
360 raise RuntimeError("No data source attached to this resource")
362 original = await self._data_source.find_one(item_id)
363 data: dict = dict(original) if isinstance(original, dict) else {}
364 if not data and hasattr(original, "__dict__"):
365 data = dict(original.__dict__)
366 data = await self.before_clone(data)
367 new_record = await self._data_source.create(data)
368 await self.after_clone(new_record)
369 return new_record
371 async def before_restore(self, data: dict) -> dict:
372 """Hook called before a soft-deleted record is restored.
374 Sets ``deleted_at`` to ``None`` by default. Override to
375 customise restore behaviour.
377 Args:
378 data: Record data dict fetched from the data source.
380 Returns:
381 Modified data dict to be passed to ``update``.
382 """
383 return {"deleted_at": None}
385 async def after_restore(self, record: Any) -> None:
386 """Hook called after a record has been restored.
388 Args:
389 record: The restored record returned by the data source.
390 """
392 async def restore(self, item_id: Any) -> Any:
393 """Restore a soft-deleted record.
395 Fetches the existing record, calls :meth:`before_restore` to
396 prepare the data, updates the record via the data source, and
397 calls :meth:`after_restore` with the result.
399 Args:
400 item_id: Identifier of the record to restore.
402 Returns:
403 The restored record.
405 Raises:
406 RuntimeError: If no data source is attached.
407 """
408 if self._data_source is None:
409 raise RuntimeError("No data source attached to this resource")
411 original = await self._data_source.find_one(item_id)
412 data: dict = dict(original) if isinstance(original, dict) else {}
413 if not data and hasattr(original, "__dict__"):
414 data = dict(original.__dict__)
415 data = await self.before_restore(data)
416 new_record = await self._data_source.update(item_id, data)
417 await self.after_restore(new_record)
418 return new_record
420 async def before_purge(self, data: dict) -> dict:
421 """Hook called before a record is permanently purged.
423 Args:
424 data: Record data dict fetched from the data source.
426 Returns:
427 Modified data dict (default: unchanged).
428 """
429 return data
431 async def after_purge(self, item_id: Any) -> None:
432 """Hook called after a record has been permanently purged.
434 Args:
435 item_id: Identifier of the purged record.
436 """
438 async def purge(self, item_id: Any) -> None:
439 """Permanently delete (purge) a record.
441 Fetches the existing record, calls :meth:`before_purge` to
442 prepare the data, hard-deletes via the data source, and calls
443 :meth:`after_purge` with the item id.
445 Args:
446 item_id: Identifier of the record to purge.
448 Raises:
449 RuntimeError: If no data source is attached.
450 """
451 if self._data_source is None:
452 raise RuntimeError("No data source attached to this resource")
454 original = await self._data_source.find_one(item_id)
455 data: dict = dict(original) if isinstance(original, dict) else {}
456 if not data and hasattr(original, "__dict__"):
457 data = dict(original.__dict__)
458 await self.before_purge(data)
459 await self._data_source.delete(item_id)
460 await self.after_purge(item_id)
462 async def before_create(self, data: dict) -> dict:
463 """Hook called before a record is created.
465 Args:
466 data: Record data to be created
468 Returns:
469 Modified data
470 """
471 return data
473 async def before_validate(self, data: dict) -> Any:
474 """Validate and coerce form data against the resource model.
476 Base implementation coerces HTML form strings to proper Python types
477 via _coerce_form_data, then validates against ``self.model``.
478 Returns Ok(coerced_data) on success, Err(AdminValidationError) with
479 per-field errors on failure.
481 Override in subclasses to add custom validation logic.
482 """
483 from lexigram.admin.exceptions import AdminValidationError
484 from lexigram.admin.resources.handler import _coerce_form_data
485 from lexigram.contracts.exceptions.domain import FieldError
486 from lexigram.result import Err, Ok
488 coerced = _coerce_form_data(data, self.model)
489 if self.model is None:
490 return Ok(coerced)
492 try:
493 self.model.model_validate(coerced)
494 except (ValueError, TypeError) as exc:
495 msg = str(exc)
496 errors: list[FieldError] = []
498 is_pydantic = (
499 type(exc).__name__ == "ValidationError"
500 and "pydantic" in type(exc).__module__
501 )
502 if is_pydantic:
503 for err in exc.errors(): # type: ignore[union-attr]
504 field = str(err["loc"][0]) if err.get("loc") else None
505 if field and field in coerced:
506 errors.append(FieldError(field=field, message=err["msg"]))
507 else:
508 field = None
509 if msg.startswith("Field '"):
510 field = msg.split("'")[1]
511 if field:
512 errors.append(FieldError(field=field, message=msg))
514 if errors:
515 return Err(
516 AdminValidationError(
517 message="Form validation failed",
518 errors=errors,
519 )
520 )
521 return Ok(coerced)
523 return Ok(coerced)
525 async def after_create(self, record: Any) -> None:
526 """Hook called after a record is created.
528 Args:
529 record: Created record
530 """
532 async def before_update(self, item_id: Any, data: dict) -> dict:
533 """Hook called before a record is updated.
535 Args:
536 item_id: Record identifier
537 data: Updated record data
539 Returns:
540 Modified data
541 """
542 return data
544 async def after_update(self, record: Any) -> None:
545 """Hook called after a record is updated.
547 Args:
548 record: Updated record
549 """
551 async def before_delete(self, item_id: Any) -> None:
552 """Hook called before a record is deleted.
554 Args:
555 item_id: Record identifier
556 """
558 async def after_delete(self, item_id: Any) -> None:
559 """Hook called after a record is deleted.
561 Args:
562 item_id: Record identifier
563 """
565 @classmethod
566 def get_table_config(cls) -> TableConfiguration:
567 """Get the table configuration for this resource.
569 Returns:
570 TableConfiguration with columns, actions, filters
571 """
572 cfg = cls.config
574 # Resolve configuration with priority: Config Object > Class Attribute
575 per_page = (
576 cls._get_config_value(cfg, "per_page", cls.page_size)
577 if cfg
578 else cls.page_size
579 )
580 default_sort = (
581 cls._get_config_value(
582 cfg,
583 "default_sort_field",
584 cls.default_sort,
585 )
586 if cfg
587 else cls.default_sort
588 )
589 default_sort_order = (
590 cls._get_config_value(cfg, "default_sort_order", "asc") if cfg else "asc"
591 )
592 action_layout = (
593 cls._get_config_value(cfg, "action_layout", cls.action_layout)
594 if cfg
595 else cls.action_layout
596 )
598 resource_name = cls.label or cls.__name__.replace("Resource", "")
599 if cfg and cfg.display_name:
600 resource_name = cfg.display_name
602 columns = list(
603 cls._get_config_value(cfg, "columns", cls.columns) if cfg else cls.columns
604 )
605 actions = list(
606 cls._get_config_value(cfg, "actions", cls.actions) if cfg else cls.actions
607 )
608 filters = list(
609 cls._get_config_value(cfg, "filters_list", cls.filters)
610 if cfg
611 else cls.filters
612 )
614 # Check class attribute `layout_type` as fallback for legacy resources
615 layout_fallback = getattr(cls, "layout_type", "stack")
616 default_layout = (
617 cls._get_config_value(cfg, "layout", layout_fallback)
618 if cfg
619 else layout_fallback
620 )
621 # Check class attribute `data_view` as fallback for legacy resources
622 view_fallback = getattr(cls, "data_view", "tabular")
623 default_view = (
624 cls._get_config_value(cfg, "view", view_fallback) if cfg else view_fallback
625 )
627 return TableConfiguration(
628 columns=columns,
629 actions=actions,
630 header_actions=[],
631 bulk_actions=list(cls.bulk_actions),
632 filter_options=filters,
633 per_page=per_page,
634 default_sort_by=default_sort,
635 default_sort_order=default_sort_order,
636 resource_name=resource_name,
637 action_layout=action_layout,
638 default_layout=default_layout,
639 default_view=default_view,
640 )
642 @classmethod
643 def get_form_class(cls) -> type[FormBase] | None:
644 """Return the Form class to use for create/edit views.
646 Returns:
647 Form class or None
648 """
649 return cls.form_class
651 @classmethod
652 def get_form_display_mode(cls) -> str:
653 """Return the form display mode for create/edit views.
655 Returns:
656 Display mode: "page", "modal", or "slider"
657 """
658 cfg = cls.config
659 return (
660 cls._get_config_value(cfg, "form_display_mode", cls.form_display_mode)
661 if cfg
662 else cls.form_display_mode
663 )
665 @classmethod
666 def get_layout_manager(cls) -> LayoutManager:
667 """Get layout manager with configured views.
669 Returns:
670 LayoutManager instance
671 """
672 from lexigram.admin.layout import LayoutManager
674 manager = LayoutManager()
675 cfg = cls.config
677 if cfg and cfg.views_list:
678 for view in cfg.views_list:
679 if hasattr(view, "to_config"):
680 layout_config = view.to_config()
681 cls._apply_layout_config(manager, layout_config)
683 # Set default view
684 if cfg.view:
685 with contextlib.suppress(ValueError):
686 manager.set_default(cfg.view)
688 return manager
690 @staticmethod
691 def _get_config_value(cfg: Any, attr: str, default: Any) -> Any:
692 """Get configuration value with fallback to default or private attribute.
694 Args:
695 cfg: Configuration object
696 attr: Attribute name
697 default: Default value if not found
699 Returns:
700 Configuration value or default
701 """
702 if cfg is None:
703 return default
705 # Try public attribute/property
706 val = getattr(cfg, attr, None)
708 # If it's the fluent method (callable) or missing, try the private attribute
709 if val is None or callable(val):
710 val = getattr(cfg, f"_{attr}", None)
712 return val if val is not None else default
714 @staticmethod
715 def _apply_layout_config(manager: LayoutManager, config: Any) -> None:
716 """Apply layout configuration to manager using configurator registry."""
717 _layout_configurator_registry.configure(manager, config)
720# Registry-style configurators for layout types ---------------------------------
721from typing import Protocol
724class LayoutConfiguratorProtocol(Protocol):
725 """Protocol for layout configurators."""
727 def can_configure(self, layout_type: Any) -> bool: ...
729 def configure_layout(self, manager: LayoutManager, config: Any) -> None: ...
732class GridLayoutConfigurator:
733 def can_configure(self, layout_type: Any) -> bool:
734 from lexigram.admin.layout import LayoutType
736 return layout_type == LayoutType.GRID
738 def configure_layout(self, manager: LayoutManager, config: Any) -> None:
739 manager.add_grid_layout( # type: ignore[attr-defined]
740 columns=config.columns,
741 card_template=config.card_template,
742 enabled=config.enabled,
743 )
746class CalendarLayoutConfigurator:
747 def can_configure(self, layout_type: Any) -> bool:
748 from lexigram.admin.layout import LayoutType
750 return layout_type == LayoutType.CALENDAR
752 def configure_layout(self, manager: LayoutManager, config: Any) -> None:
753 manager.add_calendar_layout( # type: ignore[attr-defined]
754 date_field=config.date_field,
755 title_field=config.title_field,
756 enabled=config.enabled,
757 )
760class MapLayoutConfigurator:
761 def can_configure(self, layout_type: Any) -> bool:
762 from lexigram.admin.layout import LayoutType
764 return layout_type == LayoutType.MAP
766 def configure_layout(self, manager: LayoutManager, config: Any) -> None:
767 manager.add_map_layout( # type: ignore[attr-defined]
768 latitude_field=config.latitude_field,
769 longitude_field=config.longitude_field,
770 marker_template=config.marker_template,
771 enabled=config.enabled,
772 )
775class ListLayoutConfigurator:
776 def can_configure(self, layout_type: Any) -> bool:
777 from lexigram.admin.layout import LayoutType
779 return layout_type == LayoutType.LIST
781 def configure_layout(self, manager: LayoutManager, config: Any) -> None:
782 manager.add_list_layout(enabled=config.enabled) # type: ignore[attr-defined]
785class LayoutConfiguratorRegistry:
786 """Registry that selects a configurator for a given layout type."""
788 def __init__(self) -> None:
789 self._configurators: list[LayoutConfiguratorProtocol] = [
790 GridLayoutConfigurator(),
791 CalendarLayoutConfigurator(),
792 MapLayoutConfigurator(),
793 ListLayoutConfigurator(),
794 ]
796 def configure(self, manager: LayoutManager, config: Any) -> None:
797 for c in self._configurators:
798 if c.can_configure(config.type):
799 c.configure_layout(manager, config)
800 return
801 # Fallback: raise to make misconfiguration explicit
802 raise ValueError(f"No configurator for layout type: {config.type}")
805_layout_configurator_registry = LayoutConfiguratorRegistry()
808__all__ = ["Resource"]