Coverage for src/lexigram/admin/services/resource_manager.py: 0%
164 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-24 23:18 +0800
1"""Resource manager for orchestrating CRUD with validation and authorization.
3This module provides the ResourceManager class that coordinates data access,
4validation, and authorization using the Result type for explicit error handling.
6SVC-01: ResourceManager implementation.
7"""
9from __future__ import annotations
11from typing import TYPE_CHECKING, Any, Generic, Protocol, TypeVar, runtime_checkable
13from lexigram.admin.data.paged_result import PagedResult
14from lexigram.admin.data.query import QuerySpec
15from lexigram.admin.exceptions import (
16 AdminDataError,
17 AdminValidationError,
18 NotFoundError,
19)
20from lexigram.contracts.audit import AuditEntry, AuditEventSeverity, AuditLoggerProtocol
21from lexigram.contracts.auth import AuthorizerProtocol
22from lexigram.contracts.exceptions import PermissionDeniedError as PermissionDenied
23from lexigram.di.decorators import inject
24from lexigram.result import Err, Ok, Result
26if TYPE_CHECKING:
27 from lexigram.contracts.data.sql.unit_of_work import UnitOfWorkProtocol
29T = TypeVar("T")
32@runtime_checkable
33class ResourceDataSourceProtocol(Protocol[T]):
34 """Protocol for data access operations."""
36 async def find_many(self, query: QuerySpec) -> PagedResult[T]: ...
38 async def find_one(self, item_id: Any) -> T | None: ...
40 async def create(self, data: dict[str, Any]) -> T: ...
42 async def update(self, item_id: Any, data: dict[str, Any]) -> T: ...
44 async def delete(self, item_id: Any) -> bool: ...
47@runtime_checkable
48class ResultDataSource(Protocol[T]):
49 """Enhanced protocol for data access operations that return Results.
51 This is the idiomatic pattern for new implementations.
52 Existing ResourceDataSourceProtocol implementations can be wrapped or gradually migrated
53 to this protocol.
54 """
56 async def find_many(
57 self, query: QuerySpec
58 ) -> Result[PagedResult[T], AdminDataError]: ...
60 async def find_one(self, item_id: Any) -> Result[T, AdminDataError]: ...
62 async def create(self, data: dict[str, Any]) -> Result[T, AdminDataError]: ...
64 async def update(
65 self, item_id: Any, data: dict[str, Any]
66 ) -> Result[T, AdminDataError]: ...
68 async def delete(self, item_id: Any) -> Result[bool, AdminDataError]: ...
71@runtime_checkable
72class Validator(Protocol):
73 """Protocol for data validation."""
75 async def validate(
76 self,
77 data: dict[str, Any],
78 ) -> Result[dict[str, Any], AdminValidationError]: ...
81class DefaultValidator:
82 """Default validator that accepts all data."""
84 async def validate(
85 self,
86 data: dict[str, Any],
87 ) -> Result[dict[str, Any], AdminValidationError]:
88 return Ok(data)
91class DefaultAuthorizer:
92 """Default authorizer that allows all operations."""
94 async def can_view(
95 self,
96 user: Any,
97 resource: str,
98 record: Any = None,
99 ) -> bool:
100 return True
102 async def can_create(
103 self,
104 user: Any,
105 resource: str,
106 ) -> bool:
107 return True
109 async def can_update(
110 self,
111 user: Any,
112 resource: str,
113 record: Any = None,
114 ) -> bool:
115 return True
117 async def can_delete(
118 self,
119 user: Any,
120 resource: str,
121 record: Any = None,
122 ) -> bool:
123 return True
125 async def can_execute_action(
126 self,
127 user: Any,
128 resource: str,
129 action: str,
130 record: Any | None = None,
131 ) -> bool:
132 return True
135@inject
136class ResourceManager(Generic[T]):
137 """Orchestrates CRUD operations with validation and authorization.
139 ResourceManager provides a high-level interface for resource operations
140 that combines:
141 - Data access through ResourceDataSourceProtocol protocol
142 - Input validation through Validator protocol
143 - Authorization through AuthorizerProtocol protocol
145 All operations return Result types for explicit error handling without
146 exceptions.
148 Example:
149 >>> manager = ResourceManager(
150 ... resource_name="users",
151 ... data_source=user_data_source,
152 ... validator=user_validator,
153 ... authorizer=role_authorizer,
154 ... )
155 >>> result = await manager.list(query, user=current_user)
156 >>> if result.is_ok():
157 ... users = result.unwrap()
158 ... else:
159 ... error = result.unwrap_err()
160 """
162 def __init__(
163 self,
164 resource_name: str,
165 data_source: ResourceDataSourceProtocol[T],
166 validator: Validator | None = None,
167 authorizer: AuthorizerProtocol | None = None,
168 model: type[T] | None = None,
169 uow: UnitOfWorkProtocol | None = None,
170 audit: AuditLoggerProtocol | None = None,
171 ):
172 """Initialize the resource manager.
174 Args:
175 resource_name: Name of the resource (for authorization/display)
176 data_source: Data source for CRUD operations
177 validator: Optional validator for input validation
178 authorizer: Optional authorizer for access control
179 model: Optional model class for typing results
180 uow: Optional unit-of-work for transactional bulk operations
181 audit: Optional audit logger for recording operations
182 """
183 self.resource_name = resource_name
184 self.data_source = data_source
185 self.validator = validator or DefaultValidator()
186 self.authorizer = authorizer or DefaultAuthorizer()
187 self.model = model
188 self._uow = uow
189 self._audit = audit
191 def _is_result_data_source(self) -> bool:
192 """Check if data_source implements ResultDataSource protocol.
194 Uses a marker attribute for reliable detection rather than isinstance,
195 which doesn't check return types for @runtime_checkable protocols.
196 Only returns True if explicitly marked as result-based.
197 """
198 # Check for explicit marker attribute indicating result-based adapter
199 # Default to False if not explicitly set (conservative approach)
200 return getattr(self.data_source, "returns_result", False) is True
202 async def _find_many_safe(
203 self, query: QuerySpec
204 ) -> Result[PagedResult[T], AdminDataError]:
205 """Safely call find_many, handling both ResourceDataSourceProtocol and ResultDataSource.
207 If data_source implements ResultDataSource, use it directly.
208 Otherwise, wrap the traditional ResourceDataSourceProtocol method with error handling.
209 """
210 if self._is_result_data_source():
211 return await self.data_source.find_many(query) # type: ignore[return-value]
213 try:
214 result = await self.data_source.find_many(query)
215 return Ok(result)
216 except (ConnectionError, RuntimeError, ValueError, OSError) as e:
217 return Err(AdminDataError(f"Failed to list {self.resource_name}: {e}"))
219 async def _find_one_safe(self, item_id: Any) -> Result[T | None, AdminDataError]:
220 """Safely call find_one, handling both ResourceDataSourceProtocol and ResultDataSource."""
221 if self._is_result_data_source():
222 return await self.data_source.find_one(item_id) # type: ignore[return-value]
224 try:
225 result = await self.data_source.find_one(item_id)
226 return Ok(result)
227 except (ConnectionError, RuntimeError, ValueError, OSError) as e:
228 return Err(
229 AdminDataError(f"Failed to find {self.resource_name} {item_id}: {e}")
230 )
232 async def _record_audit(
233 self,
234 *,
235 action: str,
236 actor: Any,
237 resource_id: str,
238 outcome: str,
239 severity: AuditEventSeverity,
240 **metadata: object,
241 ) -> None:
242 """Record an audit event for a resource operation."""
243 if self._audit is None:
244 return
246 actor_id = getattr(actor, "id", str(actor))
247 await self._audit.log(
248 AuditEntry(
249 action=action,
250 actor_id=actor_id,
251 resource_type=self.resource_name,
252 resource_id=resource_id,
253 outcome=outcome,
254 severity=severity,
255 metadata=dict(metadata),
256 source="admin",
257 )
258 )
260 async def list(
261 self,
262 query: QuerySpec,
263 *,
264 user: Any = None,
265 ) -> Result[PagedResult[T], PermissionDenied | AdminDataError]:
266 """List resources with authorization check.
268 Args:
269 query: Query specification for filtering/pagination
270 user: Current user for authorization
272 Returns:
273 Result containing PagedResult on success, error on failure
274 """
275 if not await self.authorizer.can_view(user, self.resource_name):
276 return Err(
277 PermissionDenied(
278 resource=self.resource_name,
279 action="view",
280 message=f"Cannot view {self.resource_name}",
281 ),
282 )
284 # Use safe wrapper that handles both ResourceDataSourceProtocol and ResultDataSource
285 find_result = await self._find_many_safe(query)
286 if find_result.is_err():
287 return Err(find_result.unwrap_err())
289 result = find_result.unwrap()
291 # Convert to PagedResult and transform to model if specified
292 paged = PagedResult(
293 items=result.items,
294 total=result.total,
295 page=result.page,
296 per_page=result.per_page,
297 cursor=getattr(result, "cursor", None),
298 )
300 if self.model:
301 paged = paged.map(lambda x: self.model(**x) if isinstance(x, dict) else x)
303 return Ok(paged)
305 async def get(
306 self,
307 item_id: Any,
308 *,
309 user: Any = None,
310 ) -> Result[T, PermissionDenied | NotFoundError | AdminDataError]:
311 """Get a single resource by ID.
313 Args:
314 item_id: Resource identifier
315 user: Current user for authorization
317 Returns:
318 Result containing resource on success, error on failure
319 """
320 # Use safe wrapper that handles both ResourceDataSourceProtocol and ResultDataSource
321 find_result = await self._find_one_safe(item_id)
322 if find_result.is_err():
323 return Err(find_result.unwrap_err())
325 record = find_result.unwrap()
327 if record is None:
328 return Err(
329 NotFoundError(
330 resource=self.resource_name,
331 identifier=str(item_id),
332 message=f"{self.resource_name} not found",
333 ),
334 )
336 if not await self.authorizer.can_view(user, self.resource_name, record):
337 return Err(
338 PermissionDenied(
339 resource=self.resource_name,
340 action="view",
341 message=f"Cannot view this {self.resource_name}",
342 ),
343 )
345 if self.model and isinstance(record, dict):
346 record = self.model(**record)
348 return Ok(record)
350 async def create(
351 self,
352 data: dict[str, Any],
353 *,
354 user: Any = None,
355 ) -> Result[T, PermissionDenied | AdminValidationError]:
356 """Create a new resource.
358 Args:
359 data: Resource data
360 user: Current user for authorization
362 Returns:
363 Result containing created resource on success, error on failure
364 """
365 if not await self.authorizer.can_create(user, self.resource_name):
366 return Err(
367 PermissionDenied(
368 resource=self.resource_name,
369 action="create",
370 message=f"Cannot create {self.resource_name}",
371 ),
372 )
374 # Validate data
375 validation_result = await self.validator.validate(data)
376 if validation_result.is_err():
377 return validation_result # type: ignore[return-value]
379 validated_data = validation_result.unwrap()
380 record = await self.data_source.create(validated_data)
382 if self.model and isinstance(record, dict):
383 record = self.model(**record)
385 # Record successful creation audit event
386 resource_id = getattr(record, "id", str(record))
387 await self._record_audit(
388 action="admin.resource.create",
389 actor=user,
390 resource_id=resource_id,
391 outcome="success",
392 severity=AuditEventSeverity.MEDIUM,
393 )
395 return Ok(record)
397 async def update(
398 self,
399 item_id: Any,
400 data: dict[str, Any],
401 *,
402 user: Any = None,
403 ) -> Result[T, PermissionDenied | AdminValidationError | NotFoundError]:
404 """Update an existing resource.
406 Args:
407 item_id: Resource identifier
408 data: Update data
409 user: Current user for authorization
411 Returns:
412 Result containing updated resource on success, error on failure
413 """
414 # Check if record exists
415 record = await self.data_source.find_one(item_id)
416 if record is None:
417 return Err(
418 NotFoundError(
419 resource=self.resource_name,
420 identifier=str(item_id),
421 message=f"{self.resource_name} not found",
422 ),
423 )
425 # Check authorization
426 if not await self.authorizer.can_update(user, self.resource_name, record):
427 return Err(
428 PermissionDenied(
429 resource=self.resource_name,
430 action="update",
431 message=f"Cannot update this {self.resource_name}",
432 ),
433 )
435 # Validate data
436 validation_result = await self.validator.validate(data)
437 if validation_result.is_err():
438 return validation_result # type: ignore[return-value]
440 validated_data = validation_result.unwrap()
441 updated = await self.data_source.update(item_id, validated_data)
443 if self.model and isinstance(updated, dict):
444 updated = self.model(**updated)
446 # Record successful update audit event
447 resource_id = getattr(updated, "id", str(item_id))
448 await self._record_audit(
449 action="admin.resource.update",
450 actor=user,
451 resource_id=resource_id,
452 outcome="success",
453 severity=AuditEventSeverity.MEDIUM,
454 )
456 return Ok(updated)
458 async def _delete_internal(
459 self,
460 item_id: Any,
461 user: Any = None,
462 *,
463 skip_audit: bool = False,
464 ) -> Result[bool, PermissionDenied | NotFoundError]:
465 """Internal delete implementation shared by delete() and bulk_delete().
467 Args:
468 item_id: Resource identifier
469 user: Current user for authorization
470 skip_audit: If True, skip audit recording (used by bulk_delete fallback)
472 Returns:
473 Result containing True on success, error on failure
474 """
475 # Check if record exists
476 record = await self.data_source.find_one(item_id)
477 if record is None:
478 return Err(
479 NotFoundError(
480 resource=self.resource_name,
481 identifier=str(item_id),
482 message=f"{self.resource_name} not found",
483 ),
484 )
486 # Check authorization
487 if not await self.authorizer.can_delete(user, self.resource_name, record):
488 return Err(
489 PermissionDenied(
490 resource=self.resource_name,
491 action="delete",
492 message=f"Cannot delete this {self.resource_name}",
493 ),
494 )
496 await self.data_source.delete(item_id)
498 # Record successful deletion audit event unless skipped (bulk_delete handles its own audit)
499 if not skip_audit:
500 await self._record_audit(
501 action="admin.resource.delete",
502 actor=user,
503 resource_id=str(item_id),
504 outcome="success",
505 severity=AuditEventSeverity.CRITICAL,
506 )
508 return Ok(True)
510 async def delete(
511 self,
512 item_id: Any,
513 *,
514 user: Any = None,
515 ) -> Result[bool, PermissionDenied | NotFoundError]:
516 """Delete a resource.
518 Args:
519 item_id: Resource identifier
520 user: Current user for authorization
522 Returns:
523 Result containing True on success, error on failure
524 """
525 return await self._delete_internal(item_id, user)
527 async def bulk_delete(
528 self,
529 ids: list[Any], # type: ignore[valid-type]
530 *,
531 user: Any = None,
532 ) -> Result[int, PermissionDenied]:
533 """Delete multiple resources.
535 Args:
536 ids: List of resource identifiers
537 user: Current user for authorization
539 Returns:
540 Result containing count of deleted resources, error on failure
541 """
542 # Check bulk authorization
543 if not await self.authorizer.can_delete(user, self.resource_name):
544 return Err(
545 PermissionDenied(
546 resource=self.resource_name,
547 action="delete",
548 message=f"Cannot delete {self.resource_name}",
549 ),
550 )
552 # Check ResourceDataSourceProtocol supports bulk operations
553 if hasattr(self.data_source, "delete_many"):
554 count = await self.data_source.delete_many(ids)
555 else:
556 # Fallback to individual deletes without per-item audit events
557 count = 0
558 if self._uow is not None:
559 async with self._uow:
560 for id_ in ids: # type: ignore[attr-defined]
561 result = await self._delete_internal(
562 id_, user=user, skip_audit=True
563 )
564 if result.is_ok():
565 count += 1
566 else:
567 for id_ in ids: # type: ignore[attr-defined]
568 result = await self._delete_internal(
569 id_, user=user, skip_audit=True
570 )
571 if result.is_ok():
572 count += 1
574 # Record single bulk deletion audit event for both paths
575 await self._record_audit(
576 action="admin.resource.bulk_delete",
577 actor=user,
578 resource_id="bulk",
579 outcome="success",
580 severity=AuditEventSeverity.CRITICAL,
581 deleted_count=count,
582 requested_ids=len(ids),
583 )
585 return Ok(count)
587 async def bulk_update(
588 self,
589 updates: list[tuple[Any, dict[str, Any]]], # type: ignore[valid-type]
590 *,
591 user: Any = None,
592 ) -> Result[int, PermissionDenied]:
593 """Update multiple resources.
595 Args:
596 updates: List of (id, data) tuples
597 user: Current user for authorization
599 Returns:
600 Result containing count of updated resources, error on failure
601 """
602 # Check bulk authorization
603 if not await self.authorizer.can_update(user, self.resource_name):
604 return Err(
605 PermissionDenied(
606 resource=self.resource_name,
607 action="update",
608 message=f"Cannot update {self.resource_name}",
609 ),
610 )
612 # Check ResourceDataSourceProtocol supports bulk operations
613 if hasattr(self.data_source, "update_many"):
614 count = await self.data_source.update_many(updates)
615 return Ok(count)
617 # Fallback to individual updates
618 count = 0
619 if self._uow is not None:
620 async with self._uow:
621 for id_, data in updates: # type: ignore[attr-defined]
622 result = await self.update(id_, data, user=user)
623 if result.is_ok():
624 count += 1
625 else:
626 for id_, data in updates: # type: ignore[attr-defined]
627 result = await self.update(id_, data, user=user)
628 if result.is_ok():
629 count += 1
631 return Ok(count)