Coverage for src/lexigram/admin/resources/base.py: 81%

307 statements  

« prev     ^ index     » next       coverage.py v7.15.4, created at 2026-08-21 15:04 +0800

1"""Base resource class for Admin Resources. 

2 

3.. stability:: stable 

4 

5Resources define the configuration for admin UI views including 

6columns, actions, filters, and permissions. 

7""" 

8 

9from __future__ import annotations 

10 

11import contextlib 

12import re 

13from typing import TYPE_CHECKING, Any 

14import warnings 

15 

16from lexigram.admin.data.data_source import IDataSource 

17from lexigram.admin.resources.config import TableConfiguration 

18 

19if TYPE_CHECKING: 

20 from lexigram.admin.actions.base import HeaderAction 

21 from lexigram.admin.forms.components import FormBase 

22 from lexigram.admin.layout.layout_manager import LayoutManager 

23 from lexigram.admin.rbac.schema import ResourcePermissions 

24 from lexigram.admin.relations.manager_ext import RelationManager 

25 from lexigram.admin.ui.filters.base import Filter 

26 from lexigram.domain import DomainModel 

27 from lexigram.ui.actions import Action, BulkAction 

28 from lexigram.ui.columns import Column 

29 

30_VALID_NAME_RE = re.compile(r"^[a-z][a-z0-9_]*(\.[a-z][a-z0-9_]*)?$") 

31 

32 

33def _validate_resource_name(name: str) -> None: 

34 """Validate a Resource name is a dotted slug. 

35 

36 Raises ``ValueError`` if the name doesn't match 

37 the slug pattern (lowercase alphanumeric with underscores, 

38 optionally dotted for namespaced resources). 

39 """ 

40 if not _VALID_NAME_RE.match(name): 

41 raise ValueError( 

42 f"Resource name {name!r} is not a valid slug. " 

43 f"Allowed: lowercase alphanumeric + underscores, " 

44 f"optionally dotted for namespaced resources. " 

45 f"Pattern: {_VALID_NAME_RE.pattern!r}" 

46 ) 

47 

48 

49class Resource: 

50 """Base class for Admin Resources. 

51 

52 Resources define the configuration for list views (tables) and form views 

53 in the admin interface. Subclass this to create custom resources. 

54 

55 Example: 

56 >>> class UserResource(Resource): 

57 ... model = UserModel 

58 ... icon = "users" 

59 ... columns = [ 

60 ... TextColumn("name").sortable(), 

61 ... TextColumn("email").sortable(), 

62 ... ] 

63 ... actions = [EditAction(), DeleteAction()] 

64 """ 

65 

66 # Data Model 

67 model: type[DomainModel] | None = None 

68 

69 # Registration metadata 

70 name: str | None = None 

71 cluster: str | None = None 

72 """Cluster name for navigation grouping. Replaces ``group``.""" 

73 

74 # Backward-compat alias for cluster 

75 group: str | None = None 

76 """Deprecated: use ``cluster`` instead. Kept in sync via __init_subclass__.""" 

77 

78 # Permissions 

79 permissions: ResourcePermissions | None = None 

80 

81 # UI Configuration 

82 icon: str = "box" 

83 label: str | None = None 

84 visible_in_sidebar: bool = True 

85 

86 # Table Configuration 

87 columns: list[Column] = [] 

88 actions: list[Action] = [] 

89 action_layout: str = "horizontal" 

90 header_actions: list[HeaderAction] = [] 

91 bulk_actions: list[BulkAction] = [] 

92 filters: list[Filter] = [] 

93 

94 # New declarative field system — SchemaField instances 

95 # When set, columns and filters are derived automatically. 

96 fields: list[Any] = [] 

97 

98 page_size: int = 20 

99 default_sort: str | None = None 

100 

101 # Optional default grouping column (users can override via the toolbar) 

102 group_by: str | None = None 

103 

104 # Table empty-state copy overrides (None = framework defaults) 

105 empty_state_title: str | None = None 

106 empty_state_message: str | None = None 

107 empty_state_icon: str | None = None 

108 

109 # Form Configuration 

110 form_class: type[FormBase] | None = None 

111 # Form display mode: "page" (full page), "modal" (centered modal), "slider" (side panel) 

112 form_display_mode: str = "modal" # Options: "page", "modal", "slider" 

113 # Model fields excluded from generated forms (e.g. secrets, framework-managed 

114 # columns). Resources may extend the default to exclude their own fields. 

115 form_exclude_fields: tuple[str, ...] = ("id", "created_at", "updated_at") 

116 

117 # Resource Config (Optional fluent config) 

118 config: Any = None 

119 

120 def __init_subclass__(cls, **kwargs: Any) -> None: 

121 """Validate and auto-derive backward-compat attributes when using ``fields``.""" 

122 super().__init_subclass__(**kwargs) 

123 

124 own = cls.__dict__ 

125 

126 # Sync group <-> cluster for backward compatibility 

127 if "group" in own and "cluster" not in own: 

128 cls.cluster = own["group"] 

129 if "cluster" in own and "group" not in own: 

130 cls.group = own["cluster"] 

131 

132 # Validate name if explicitly set — must be a dotted slug 

133 if "name" in own and own["name"] is not None: 

134 _validate_resource_name(own["name"]) 

135 

136 has_fields = "fields" in own 

137 has_columns = "columns" in own 

138 has_filters = "filters" in own 

139 has_form_class = "form_class" in own 

140 

141 if has_fields: 

142 if has_columns or has_filters or has_form_class: 

143 warnings.warn( 

144 "Resource.fields is the new declarative path for schema " 

145 "configuration. When fields is set, columns, filters, and " 

146 "form_class should not be set — they will be derived from " 

147 "fields automatically.", 

148 DeprecationWarning, 

149 stacklevel=2, 

150 ) 

151 

152 # Derive columns from fields for backward compatibility 

153 if not has_columns: 

154 cls.columns = list(cls.fields) 

155 

156 # Derive filters from fields for backward compatibility 

157 if not has_filters: 

158 cls.filters = [f for f in cls.fields if getattr(f, "filterable", False)] 

159 

160 # Relation managers for inline related-record editing on the ViewPage 

161 relations: list[type[RelationManager]] = [] 

162 

163 # Search Configuration 

164 # Fields to include in global search queries. Empty list disables search for this resource. 

165 search_fields: list[str] = [] 

166 # Field used as the display title in search results (falls back to "id") 

167 search_title_field: str = "name" 

168 

169 # Optional integration knobs 

170 cacheable: bool | Any = False # True or CacheableSpec enables list caching 

171 searchable: bool | Any = False # True or SearchableSpec enables search index 

172 resilient: bool | Any = False # True or ResilientSpec enables retry/circuit 

173 

174 def cache_spec(self) -> Any: 

175 """Return a CacheableSpec or None based on the cacheable field.""" 

176 if self.cacheable is False: 

177 return None 

178 if self.cacheable is True: 

179 from lexigram.admin.integrations.cache import CacheableSpec 

180 

181 return CacheableSpec() 

182 return self.cacheable 

183 

184 def search_spec(self) -> Any: 

185 """Return a SearchableSpec or None based on the searchable field.""" 

186 if self.searchable is False: 

187 return None 

188 if self.searchable is True: 

189 from lexigram.contracts.search import SearchableSpec 

190 

191 return SearchableSpec( 

192 index_name=self.name, 

193 fields=tuple(self.search_fields), 

194 ) 

195 return self.searchable 

196 

197 def resilient_spec(self) -> Any: 

198 """Return a ResilientSpec or None based on the resilient field.""" 

199 if self.resilient is False: 

200 return None 

201 if self.resilient is True: 

202 from lexigram.admin.integrations.resilience import ResilientSpec 

203 

204 return ResilientSpec() 

205 return self.resilient 

206 

207 # Data source instance for search (set at runtime via set_data_source) 

208 _data_source: IDataSource | None = None 

209 

210 def set_data_source(self, data_source: IDataSource) -> None: 

211 """Attach a data source to this resource for search and list support. 

212 

213 Args: 

214 data_source: An IDataSource-compatible instance. 

215 

216 Raises: 

217 TypeError: If data_source does not satisfy IDataSource protocol. 

218 """ 

219 if not isinstance(data_source, IDataSource): 

220 raise TypeError( 

221 f"data_source must implement IDataSource, got {type(data_source).__name__}" 

222 ) 

223 self._data_source = data_source 

224 

225 async def search(self, query: str, *, limit: int = 5) -> list[dict[str, Any]]: 

226 """Search this resource for items matching *query*. 

227 

228 Override in subclasses for custom search logic. The default 

229 implementation queries the attached data source using 

230 :attr:`search_fields`. 

231 

232 Args: 

233 query: Search term entered by the user. 

234 limit: Maximum number of results to return. 

235 

236 Returns: 

237 List of dicts with ``id``, ``title``, and ``subtitle`` keys. 

238 """ 

239 if not self.search_fields or self._data_source is None: 

240 return [] 

241 

242 from lexigram.admin.data.query import QuerySpec 

243 

244 qs = ( 

245 QuerySpec() 

246 .with_search(query, self.search_fields) 

247 .with_page(1) 

248 .with_per_page(limit) 

249 ) 

250 try: 

251 result = await self._data_source.find_many(qs) 

252 except (AttributeError, TypeError, ValueError, KeyError, RuntimeError): 

253 return [] 

254 

255 hits: list[dict[str, Any]] = [] 

256 for item in result.items: 

257 if isinstance(item, dict): 

258 item_id = item.get("id", "") 

259 title = ( 

260 item.get(self.search_title_field) 

261 or item.get("name") 

262 or item.get("title") 

263 or str(item_id) 

264 ) 

265 subtitle = item.get("email") or item.get("description") or "" 

266 else: 

267 item_id = getattr(item, "id", "") 

268 title = ( 

269 getattr(item, self.search_title_field, None) 

270 or getattr(item, "name", None) 

271 or str(item_id) 

272 ) 

273 subtitle = ( 

274 getattr(item, "email", "") or getattr(item, "description", "") or "" 

275 ) 

276 hits.append( 

277 {"id": str(item_id), "title": str(title), "subtitle": str(subtitle)} 

278 ) 

279 return hits 

280 

281 async def fetch_list( 

282 self, 

283 *, 

284 limit: int = 20, 

285 offset: int = 0, 

286 filters: dict[str, Any] | None = None, 

287 search: str | None = None, 

288 search_fields: list[str] | None = None, 

289 sort_by: str | None = None, 

290 sort_order: str = "asc", 

291 include_deleted: bool = False, 

292 ) -> tuple[list[Any], int]: 

293 """Fetch a paginated list of items via the attached IDataSource. 

294 

295 Builds a Query object from the pagination/search/filter/sort parameters 

296 and delegates to ``self._data_source.find_many(query)``. 

297 

298 Override this in resource subclasses for custom data access logic. 

299 

300 Returns: 

301 Tuple of (items, total_count). 

302 """ 

303 if self._data_source is None: 

304 return [], 0 

305 

306 from lexigram.admin.data.query import QuerySpec 

307 

308 page = (offset // limit) + 1 if limit else 1 

309 qs = QuerySpec().with_page(page).with_per_page(limit) 

310 

311 if search and search_fields: 

312 qs = qs.with_search(search, search_fields) 

313 

314 if sort_by: 

315 qs = qs.with_order_by(sort_by, sort_order) 

316 

317 if include_deleted: 

318 qs = qs.with_deleted(True) 

319 

320 for field, value in (filters or {}).items(): 

321 if isinstance(value, list): 

322 qs = qs.with_where_in(field, value) 

323 else: 

324 qs = qs.with_where_eq(field, value) 

325 

326 result = await self._data_source.find_many(qs) 

327 items = list(result.items) 

328 total = result.total 

329 return items, total 

330 

331 async def before_clone(self, data: dict) -> dict: 

332 """Hook called before a record is cloned. 

333 

334 Strips the ``id`` field (so a new ID is assigned) and 

335 appends `` (Copy)`` to the ``name`` field. Override 

336 to customise clone behaviour. 

337 

338 Args: 

339 data: Record data dict fetched from the data source. 

340 

341 Returns: 

342 Modified data dict to be passed to ``create``. 

343 """ 

344 data.pop("id", None) 

345 if "name" in data: 

346 data["name"] = f"{data['name']} (Copy)" 

347 return data 

348 

349 async def after_clone(self, record: Any) -> None: 

350 """Hook called after a record has been cloned. 

351 

352 Args: 

353 record: The newly created record returned by the data source. 

354 """ 

355 

356 async def duplicate(self, item_id: Any) -> Any: 

357 """Duplicate (clone) a record by its identifier. 

358 

359 Fetches the existing record via the attached data source, 

360 calls :meth:`before_clone` to prepare the data, creates 

361 a new record, and calls :meth:`after_clone` with the result. 

362 

363 Args: 

364 item_id: Identifier of the record to clone. 

365 

366 Returns: 

367 The newly created record. 

368 

369 Raises: 

370 RuntimeError: If no data source is attached. 

371 """ 

372 if self._data_source is None: 

373 raise RuntimeError("No data source attached to this resource") 

374 

375 original = await self._data_source.find_one(item_id) 

376 data: dict = dict(original) if isinstance(original, dict) else {} 

377 if not data and hasattr(original, "__dict__"): 

378 data = dict(original.__dict__) 

379 data = await self.before_clone(data) 

380 new_record = await self._data_source.create(data) 

381 await self.after_clone(new_record) 

382 return new_record 

383 

384 async def before_restore(self, data: dict) -> dict: 

385 """Hook called before a soft-deleted record is restored. 

386 

387 Sets ``deleted_at`` to ``None`` by default. Override to 

388 customise restore behaviour. 

389 

390 Args: 

391 data: Record data dict fetched from the data source. 

392 

393 Returns: 

394 Modified data dict to be passed to ``update``. 

395 """ 

396 return {"deleted_at": None} 

397 

398 async def after_restore(self, record: Any) -> None: 

399 """Hook called after a record has been restored. 

400 

401 Args: 

402 record: The restored record returned by the data source. 

403 """ 

404 

405 async def restore(self, item_id: Any) -> Any: 

406 """Restore a soft-deleted record. 

407 

408 Fetches the existing record, calls :meth:`before_restore` to 

409 prepare the data, updates the record via the data source, and 

410 calls :meth:`after_restore` with the result. 

411 

412 Args: 

413 item_id: Identifier of the record to restore. 

414 

415 Returns: 

416 The restored record. 

417 

418 Raises: 

419 RuntimeError: If no data source is attached. 

420 """ 

421 if self._data_source is None: 

422 raise RuntimeError("No data source attached to this resource") 

423 

424 original = await self._data_source.find_one(item_id) 

425 data: dict = dict(original) if isinstance(original, dict) else {} 

426 if not data and hasattr(original, "__dict__"): 

427 data = dict(original.__dict__) 

428 data = await self.before_restore(data) 

429 new_record = await self._data_source.update(item_id, data) 

430 await self.after_restore(new_record) 

431 return new_record 

432 

433 async def before_purge(self, data: dict) -> dict: 

434 """Hook called before a record is permanently purged. 

435 

436 Args: 

437 data: Record data dict fetched from the data source. 

438 

439 Returns: 

440 Modified data dict (default: unchanged). 

441 """ 

442 return data 

443 

444 async def after_purge(self, item_id: Any) -> None: 

445 """Hook called after a record has been permanently purged. 

446 

447 Args: 

448 item_id: Identifier of the purged record. 

449 """ 

450 

451 async def purge(self, item_id: Any) -> None: 

452 """Permanently delete (purge) a record. 

453 

454 Fetches the existing record, calls :meth:`before_purge` to 

455 prepare the data, hard-deletes via the data source, and calls 

456 :meth:`after_purge` with the item id. 

457 

458 Args: 

459 item_id: Identifier of the record to purge. 

460 

461 Raises: 

462 RuntimeError: If no data source is attached. 

463 """ 

464 if self._data_source is None: 

465 raise RuntimeError("No data source attached to this resource") 

466 

467 original = await self._data_source.find_one(item_id) 

468 data: dict = dict(original) if isinstance(original, dict) else {} 

469 if not data and hasattr(original, "__dict__"): 

470 data = dict(original.__dict__) 

471 await self.before_purge(data) 

472 await self._data_source.delete(item_id) 

473 await self.after_purge(item_id) 

474 

475 async def before_create(self, data: dict) -> dict: 

476 """Hook called before a record is created. 

477 

478 Args: 

479 data: Record data to be created 

480 

481 Returns: 

482 Modified data 

483 """ 

484 return data 

485 

486 async def before_validate(self, data: dict) -> Any: 

487 """Validate and coerce form data against the resource model. 

488 

489 Base implementation coerces HTML form strings to proper Python types 

490 via _coerce_form_data, then validates against ``self.model``. 

491 Returns Ok(coerced_data) on success, Err(AdminValidationError) with 

492 per-field errors on failure. 

493 

494 Override in subclasses to add custom validation logic. 

495 """ 

496 from lexigram.admin.exceptions import AdminValidationError 

497 from lexigram.admin.resources.handler import _coerce_form_data 

498 from lexigram.contracts.exceptions.domain import FieldError 

499 from lexigram.result import Err, Ok 

500 

501 coerced = _coerce_form_data(data, self.model) 

502 if self.model is None: 

503 return Ok(coerced) 

504 

505 if not hasattr(self.model, "model_validate"): 

506 return Ok(coerced) 

507 

508 try: 

509 self.model.model_validate(coerced) 

510 except (ValueError, TypeError) as exc: 

511 msg = str(exc) 

512 errors: list[FieldError] = [] 

513 

514 is_pydantic = ( 

515 type(exc).__name__ == "ValidationError" 

516 and "pydantic" in type(exc).__module__ 

517 ) 

518 if is_pydantic: 

519 for err in exc.errors(): # type: ignore[union-attr] 

520 field = str(err["loc"][0]) if err.get("loc") else None 

521 if field and field in coerced: 

522 errors.append(FieldError(field=field, message=err["msg"])) 

523 else: 

524 field = None 

525 if msg.startswith("Field '"): 

526 field = msg.split("'")[1] 

527 if field: 

528 errors.append(FieldError(field=field, message=msg)) 

529 

530 if errors: 

531 return Err( 

532 AdminValidationError( 

533 message="Form validation failed", 

534 errors=errors, 

535 ) 

536 ) 

537 return Ok(coerced) 

538 

539 return Ok(coerced) 

540 

541 async def after_create(self, record: Any) -> None: 

542 """Hook called after a record is created. 

543 

544 Args: 

545 record: Created record 

546 """ 

547 

548 async def before_update(self, item_id: Any, data: dict) -> dict: 

549 """Hook called before a record is updated. 

550 

551 Args: 

552 item_id: Record identifier 

553 data: Updated record data 

554 

555 Returns: 

556 Modified data 

557 """ 

558 return data 

559 

560 async def after_update(self, record: Any) -> None: 

561 """Hook called after a record is updated. 

562 

563 Args: 

564 record: Updated record 

565 """ 

566 

567 async def before_delete(self, item_id: Any) -> None: 

568 """Hook called before a record is deleted. 

569 

570 Args: 

571 item_id: Record identifier 

572 """ 

573 

574 async def after_delete(self, item_id: Any) -> None: 

575 """Hook called after a record is deleted. 

576 

577 Args: 

578 item_id: Record identifier 

579 """ 

580 

581 @classmethod 

582 def get_action_hooks(cls, action_name: str) -> list[Any]: 

583 """Get action lifecycle hooks for the named action. 

584 

585 Override in a resource subclass to attach ``ActionHookProtocol`` 

586 hooks to registry-based actions. Hooks are collected by 

587 ``ActionExecutor`` and run before/after the action body and on 

588 failure. 

589 

590 Args: 

591 action_name: Name of the action (e.g. ``"export"``) 

592 

593 Returns: 

594 List of action hooks for the action. 

595 """ 

596 return [] 

597 

598 @classmethod 

599 def get_table_config(cls) -> TableConfiguration: 

600 """Get the table configuration for this resource. 

601 

602 Returns: 

603 TableConfiguration with columns, actions, filters 

604 """ 

605 cfg = cls.config 

606 

607 # Resolve configuration with priority: Config Object > Class Attribute 

608 per_page = ( 

609 cls._get_config_value(cfg, "per_page", cls.page_size) 

610 if cfg 

611 else cls.page_size 

612 ) 

613 default_sort = ( 

614 cls._get_config_value( 

615 cfg, 

616 "default_sort_field", 

617 cls.default_sort, 

618 ) 

619 if cfg 

620 else cls.default_sort 

621 ) 

622 default_sort_order = ( 

623 cls._get_config_value(cfg, "default_sort_order", "asc") if cfg else "asc" 

624 ) 

625 action_layout = ( 

626 cls._get_config_value(cfg, "action_layout", cls.action_layout) 

627 if cfg 

628 else cls.action_layout 

629 ) 

630 

631 resource_name = cls.label or cls.__name__.replace("Resource", "") 

632 if cfg and cfg.display_name: 

633 resource_name = cfg.display_name 

634 

635 columns = list( 

636 cls._get_config_value(cfg, "columns", cls.columns) if cfg else cls.columns 

637 ) 

638 actions = list( 

639 cls._get_config_value(cfg, "actions", cls.actions) if cfg else cls.actions 

640 ) 

641 filters = list( 

642 cls._get_config_value(cfg, "filters_list", cls.filters) 

643 if cfg 

644 else cls.filters 

645 ) 

646 

647 # Check class attribute `layout_type` as fallback for legacy resources 

648 layout_fallback = getattr(cls, "layout_type", "stack") 

649 default_layout = ( 

650 cls._get_config_value(cfg, "layout", layout_fallback) 

651 if cfg 

652 else layout_fallback 

653 ) 

654 # Check class attribute `data_view` as fallback for legacy resources 

655 view_fallback = getattr(cls, "data_view", "tabular") 

656 default_view = ( 

657 cls._get_config_value(cfg, "view", view_fallback) if cfg else view_fallback 

658 ) 

659 

660 empty_state_title = ( 

661 cls._get_config_value(cfg, "empty_state_title", cls.empty_state_title) 

662 if cfg 

663 else cls.empty_state_title 

664 ) 

665 empty_state_message = ( 

666 cls._get_config_value(cfg, "empty_state_message", cls.empty_state_message) 

667 if cfg 

668 else cls.empty_state_message 

669 ) 

670 empty_state_icon = ( 

671 cls._get_config_value(cfg, "empty_state_icon", cls.empty_state_icon) 

672 if cfg 

673 else cls.empty_state_icon 

674 ) 

675 group_by = ( 

676 cls._get_config_value(cfg, "group_by", cls.group_by) 

677 if cfg 

678 else cls.group_by 

679 ) 

680 

681 return TableConfiguration( 

682 columns=columns, 

683 actions=actions, 

684 header_actions=list(cls.header_actions), 

685 bulk_actions=list(cls.bulk_actions), 

686 filter_options=filters, 

687 per_page=per_page, 

688 default_sort_by=default_sort, 

689 default_sort_order=default_sort_order, 

690 resource_name=resource_name, 

691 action_layout=action_layout, 

692 default_layout=default_layout, 

693 default_view=default_view, 

694 empty_state_title=empty_state_title, 

695 empty_state_message=empty_state_message, 

696 empty_state_icon=empty_state_icon, 

697 group_by=group_by, 

698 ) 

699 

700 @classmethod 

701 def get_form_class(cls) -> type[FormBase] | None: 

702 """Return the Form class to use for create/edit views. 

703 

704 Returns: 

705 Form class or None 

706 """ 

707 return cls.form_class 

708 

709 @classmethod 

710 def get_form_display_mode(cls) -> str: 

711 """Return the form display mode for create/edit views. 

712 

713 Returns: 

714 Display mode: "page", "modal", or "slider" 

715 """ 

716 cfg = cls.config 

717 return ( 

718 cls._get_config_value(cfg, "form_display_mode", cls.form_display_mode) 

719 if cfg 

720 else cls.form_display_mode 

721 ) 

722 

723 @classmethod 

724 def get_layout_manager(cls) -> LayoutManager: 

725 """Get layout manager with configured views. 

726 

727 Returns: 

728 LayoutManager instance 

729 """ 

730 from lexigram.admin.layout import LayoutManager 

731 

732 manager = LayoutManager() 

733 cfg = cls.config 

734 

735 if cfg and cfg.views_list: 

736 for view in cfg.views_list: 

737 if hasattr(view, "to_config"): 

738 layout_config = view.to_config() 

739 cls._apply_layout_config(manager, layout_config) 

740 

741 # Set default view 

742 if cfg.view: 

743 with contextlib.suppress(ValueError): 

744 manager.set_default(cfg.view) 

745 

746 return manager 

747 

748 @staticmethod 

749 def _get_config_value(cfg: Any, attr: str, default: Any) -> Any: 

750 """Get configuration value with fallback to default or private attribute. 

751 

752 Args: 

753 cfg: Configuration object 

754 attr: Attribute name 

755 default: Default value if not found 

756 

757 Returns: 

758 Configuration value or default 

759 """ 

760 if cfg is None: 

761 return default 

762 

763 # Try public attribute/property 

764 val = getattr(cfg, attr, None) 

765 

766 # If it's the fluent method (callable) or missing, try the private attribute 

767 if val is None or callable(val): 

768 val = getattr(cfg, f"_{attr}", None) 

769 

770 return val if val is not None else default 

771 

772 @staticmethod 

773 def _apply_layout_config(manager: LayoutManager, config: Any) -> None: 

774 """Apply layout configuration to manager using configurator registry.""" 

775 _layout_configurator_registry.configure(manager, config) 

776 

777 

778# Registry-style configurators for layout types --------------------------------- 

779from typing import Protocol 

780 

781 

782class LayoutConfiguratorProtocol(Protocol): 

783 """Protocol for layout configurators.""" 

784 

785 def can_configure(self, layout_type: Any) -> bool: ... 

786 

787 def configure_layout(self, manager: LayoutManager, config: Any) -> None: ... 

788 

789 

790class GridLayoutConfigurator: 

791 def can_configure(self, layout_type: Any) -> bool: 

792 from lexigram.admin.layout import LayoutType 

793 

794 return layout_type == LayoutType.GRID 

795 

796 def configure_layout(self, manager: LayoutManager, config: Any) -> None: 

797 manager.add_grid_layout( # type: ignore[attr-defined] 

798 columns=config.columns, 

799 card_template=config.card_template, 

800 enabled=config.enabled, 

801 ) 

802 

803 

804class CalendarLayoutConfigurator: 

805 def can_configure(self, layout_type: Any) -> bool: 

806 from lexigram.admin.layout import LayoutType 

807 

808 return layout_type == LayoutType.CALENDAR 

809 

810 def configure_layout(self, manager: LayoutManager, config: Any) -> None: 

811 manager.add_calendar_layout( # type: ignore[attr-defined] 

812 date_field=config.date_field, 

813 title_field=config.title_field, 

814 enabled=config.enabled, 

815 ) 

816 

817 

818class MapLayoutConfigurator: 

819 def can_configure(self, layout_type: Any) -> bool: 

820 from lexigram.admin.layout import LayoutType 

821 

822 return layout_type == LayoutType.MAP 

823 

824 def configure_layout(self, manager: LayoutManager, config: Any) -> None: 

825 manager.add_map_layout( # type: ignore[attr-defined] 

826 latitude_field=config.latitude_field, 

827 longitude_field=config.longitude_field, 

828 marker_template=config.marker_template, 

829 enabled=config.enabled, 

830 ) 

831 

832 

833class ListLayoutConfigurator: 

834 def can_configure(self, layout_type: Any) -> bool: 

835 from lexigram.admin.layout import LayoutType 

836 

837 return layout_type == LayoutType.LIST 

838 

839 def configure_layout(self, manager: LayoutManager, config: Any) -> None: 

840 manager.add_list_layout(enabled=config.enabled) # type: ignore[attr-defined] 

841 

842 

843class LayoutConfiguratorRegistry: 

844 """Registry that selects a configurator for a given layout type.""" 

845 

846 def __init__(self) -> None: 

847 self._configurators: list[LayoutConfiguratorProtocol] = [ 

848 GridLayoutConfigurator(), 

849 CalendarLayoutConfigurator(), 

850 MapLayoutConfigurator(), 

851 ListLayoutConfigurator(), 

852 ] 

853 

854 def configure(self, manager: LayoutManager, config: Any) -> None: 

855 for c in self._configurators: 

856 if c.can_configure(config.type): 

857 c.configure_layout(manager, config) 

858 return 

859 # Fallback: raise to make misconfiguration explicit 

860 raise ValueError(f"No configurator for layout type: {config.type}") 

861 

862 

863_layout_configurator_registry = LayoutConfiguratorRegistry() 

864 

865 

866__all__ = ["Resource"]