Coverage for src/lexigram/admin/services/resource_manager.py: 74%
163 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 15:04 +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.query import PagedResult, QuerySpec
14from lexigram.admin.exceptions import (
15 AdminDataError,
16 AdminValidationError,
17 NotFoundError,
18)
19from lexigram.contracts.audit import AuditEntry, AuditEventSeverity, AuditLoggerProtocol
20from lexigram.contracts.auth import AuthorizerProtocol
21from lexigram.contracts.exceptions import PermissionDeniedError as PermissionDenied
22from lexigram.di.decorators import inject
23from lexigram.result import Err, Ok, Result
25if TYPE_CHECKING:
26 from lexigram.contracts.data.sql.unit_of_work import UnitOfWorkProtocol
28T = TypeVar("T")
31@runtime_checkable
32class ResourceDataSourceProtocol(Protocol[T]):
33 """Protocol for data access operations."""
35 async def find_many(self, query: QuerySpec) -> PagedResult[T]: ...
37 async def find_one(self, item_id: Any) -> T | None: ...
39 async def create(self, data: dict[str, Any]) -> T: ...
41 async def update(self, item_id: Any, data: dict[str, Any]) -> T: ...
43 async def delete(self, item_id: Any) -> bool: ...
46@runtime_checkable
47class ResultDataSource(Protocol[T]):
48 """Enhanced protocol for data access operations that return Results.
50 This is the idiomatic pattern for new implementations.
51 Existing ResourceDataSourceProtocol implementations can be wrapped or gradually migrated
52 to this protocol.
53 """
55 async def find_many(
56 self, query: QuerySpec
57 ) -> Result[PagedResult[T], AdminDataError]: ...
59 async def find_one(self, item_id: Any) -> Result[T, AdminDataError]: ...
61 async def create(self, data: dict[str, Any]) -> Result[T, AdminDataError]: ...
63 async def update(
64 self, item_id: Any, data: dict[str, Any]
65 ) -> Result[T, AdminDataError]: ...
67 async def delete(self, item_id: Any) -> Result[bool, AdminDataError]: ...
70@runtime_checkable
71class Validator(Protocol):
72 """Protocol for data validation."""
74 async def validate(
75 self,
76 data: dict[str, Any],
77 ) -> Result[dict[str, Any], AdminValidationError]: ...
80class DefaultValidator:
81 """Default validator that accepts all data."""
83 async def validate(
84 self,
85 data: dict[str, Any],
86 ) -> Result[dict[str, Any], AdminValidationError]:
87 return Ok(data)
90class DefaultAuthorizer:
91 """Default authorizer that allows all operations."""
93 async def can_view(
94 self,
95 user: Any,
96 resource: str,
97 record: Any = None,
98 ) -> bool:
99 return True
101 async def can_create(
102 self,
103 user: Any,
104 resource: str,
105 ) -> bool:
106 return True
108 async def can_update(
109 self,
110 user: Any,
111 resource: str,
112 record: Any = None,
113 ) -> bool:
114 return True
116 async def can_delete(
117 self,
118 user: Any,
119 resource: str,
120 record: Any = None,
121 ) -> bool:
122 return True
124 async def can_execute_action(
125 self,
126 user: Any,
127 resource: str,
128 action: str,
129 record: Any | None = None,
130 ) -> bool:
131 return True
134@inject
135class ResourceManager(Generic[T]):
136 """Orchestrates CRUD operations with validation and authorization.
138 ResourceManager provides a high-level interface for resource operations
139 that combines:
140 - Data access through ResourceDataSourceProtocol protocol
141 - Input validation through Validator protocol
142 - Authorization through AuthorizerProtocol protocol
144 All operations return Result types for explicit error handling without
145 exceptions.
147 Example:
148 >>> manager = ResourceManager(
149 ... resource_name="users",
150 ... data_source=user_data_source,
151 ... validator=user_validator,
152 ... authorizer=role_authorizer,
153 ... )
154 >>> result = await manager.list(query, user=current_user)
155 >>> if result.is_ok():
156 ... users = result.unwrap()
157 ... else:
158 ... error = result.unwrap_err()
159 """
161 def __init__(
162 self,
163 resource_name: str,
164 data_source: ResourceDataSourceProtocol[T],
165 validator: Validator | None = None,
166 authorizer: AuthorizerProtocol | None = None,
167 model: type[T] | None = None,
168 uow: UnitOfWorkProtocol | None = None,
169 audit: AuditLoggerProtocol | None = None,
170 ):
171 """Initialize the resource manager.
173 Args:
174 resource_name: Name of the resource (for authorization/display)
175 data_source: Data source for CRUD operations
176 validator: Optional validator for input validation
177 authorizer: Optional authorizer for access control
178 model: Optional model class for typing results
179 uow: Optional unit-of-work for transactional bulk operations
180 audit: Optional audit logger for recording operations
181 """
182 self.resource_name = resource_name
183 self.data_source = data_source
184 self.validator = validator or DefaultValidator()
185 self.authorizer = authorizer or DefaultAuthorizer()
186 self.model = model
187 self._uow = uow
188 self._audit = audit
190 def _is_result_data_source(self) -> bool:
191 """Check if data_source implements ResultDataSource protocol.
193 Uses a marker attribute for reliable detection rather than isinstance,
194 which doesn't check return types for @runtime_checkable protocols.
195 Only returns True if explicitly marked as result-based.
196 """
197 # Check for explicit marker attribute indicating result-based adapter
198 # Default to False if not explicitly set (conservative approach)
199 return getattr(self.data_source, "returns_result", False) is True
201 async def _find_many_safe(
202 self, query: QuerySpec
203 ) -> Result[PagedResult[T], AdminDataError]:
204 """Safely call find_many, handling both ResourceDataSourceProtocol and ResultDataSource.
206 If data_source implements ResultDataSource, use it directly.
207 Otherwise, wrap the traditional ResourceDataSourceProtocol method with error handling.
208 """
209 if self._is_result_data_source():
210 return await self.data_source.find_many(query) # type: ignore[return-value]
212 try:
213 result = await self.data_source.find_many(query)
214 return Ok(result)
215 except (ConnectionError, RuntimeError, ValueError, OSError) as e:
216 return Err(AdminDataError(f"Failed to list {self.resource_name}: {e}"))
218 async def _find_one_safe(self, item_id: Any) -> Result[T | None, AdminDataError]:
219 """Safely call find_one, handling both ResourceDataSourceProtocol and ResultDataSource."""
220 if self._is_result_data_source():
221 return await self.data_source.find_one(item_id) # type: ignore[return-value]
223 try:
224 result = await self.data_source.find_one(item_id)
225 return Ok(result)
226 except (ConnectionError, RuntimeError, ValueError, OSError) as e:
227 return Err(
228 AdminDataError(f"Failed to find {self.resource_name} {item_id}: {e}")
229 )
231 async def _record_audit(
232 self,
233 *,
234 action: str,
235 actor: Any,
236 resource_id: str,
237 outcome: str,
238 severity: AuditEventSeverity,
239 **metadata: object,
240 ) -> None:
241 """Record an audit event for a resource operation."""
242 if self._audit is None:
243 return
245 actor_id = getattr(actor, "id", str(actor))
246 await self._audit.log(
247 AuditEntry(
248 action=action,
249 actor_id=actor_id,
250 resource_type=self.resource_name,
251 resource_id=resource_id,
252 outcome=outcome,
253 severity=severity,
254 metadata=dict(metadata),
255 source="admin",
256 )
257 )
259 async def list(
260 self,
261 query: QuerySpec,
262 *,
263 user: Any = None,
264 ) -> Result[PagedResult[T], PermissionDenied | AdminDataError]:
265 """List resources with authorization check.
267 Args:
268 query: Query specification for filtering/pagination
269 user: Current user for authorization
271 Returns:
272 Result containing PagedResult on success, error on failure
273 """
274 if not await self.authorizer.can_view(user, self.resource_name):
275 return Err(
276 PermissionDenied(
277 resource=self.resource_name,
278 action="view",
279 message=f"Cannot view {self.resource_name}",
280 ),
281 )
283 # Use safe wrapper that handles both ResourceDataSourceProtocol and ResultDataSource
284 find_result = await self._find_many_safe(query)
285 if find_result.is_err():
286 return Err(find_result.unwrap_err())
288 result = find_result.unwrap()
290 # Convert to PagedResult and transform to model if specified
291 paged = PagedResult(
292 items=result.items,
293 total=result.total,
294 page=result.page,
295 per_page=result.per_page,
296 cursor=getattr(result, "cursor", None),
297 )
299 if self.model:
300 paged = paged.map(lambda x: self.model(**x) if isinstance(x, dict) else x)
302 return Ok(paged)
304 async def get(
305 self,
306 item_id: Any,
307 *,
308 user: Any = None,
309 ) -> Result[T, PermissionDenied | NotFoundError | AdminDataError]:
310 """Get a single resource by ID.
312 Args:
313 item_id: Resource identifier
314 user: Current user for authorization
316 Returns:
317 Result containing resource on success, error on failure
318 """
319 # Use safe wrapper that handles both ResourceDataSourceProtocol and ResultDataSource
320 find_result = await self._find_one_safe(item_id)
321 if find_result.is_err():
322 return Err(find_result.unwrap_err())
324 record = find_result.unwrap()
326 if record is None:
327 return Err(
328 NotFoundError(
329 resource=self.resource_name,
330 identifier=str(item_id),
331 message=f"{self.resource_name} not found",
332 ),
333 )
335 if not await self.authorizer.can_view(user, self.resource_name, record):
336 return Err(
337 PermissionDenied(
338 resource=self.resource_name,
339 action="view",
340 message=f"Cannot view this {self.resource_name}",
341 ),
342 )
344 if self.model and isinstance(record, dict):
345 record = self.model(**record)
347 return Ok(record)
349 async def create(
350 self,
351 data: dict[str, Any],
352 *,
353 user: Any = None,
354 ) -> Result[T, PermissionDenied | AdminValidationError]:
355 """Create a new resource.
357 Args:
358 data: Resource data
359 user: Current user for authorization
361 Returns:
362 Result containing created resource on success, error on failure
363 """
364 if not await self.authorizer.can_create(user, self.resource_name):
365 return Err(
366 PermissionDenied(
367 resource=self.resource_name,
368 action="create",
369 message=f"Cannot create {self.resource_name}",
370 ),
371 )
373 # Validate data
374 validation_result = await self.validator.validate(data)
375 if validation_result.is_err():
376 return validation_result # type: ignore[return-value]
378 validated_data = validation_result.unwrap()
379 record = await self.data_source.create(validated_data)
381 if self.model and isinstance(record, dict):
382 record = self.model(**record)
384 # Record successful creation audit event
385 resource_id = getattr(record, "id", str(record))
386 await self._record_audit(
387 action="admin.resource.create",
388 actor=user,
389 resource_id=resource_id,
390 outcome="success",
391 severity=AuditEventSeverity.MEDIUM,
392 )
394 return Ok(record)
396 async def update(
397 self,
398 item_id: Any,
399 data: dict[str, Any],
400 *,
401 user: Any = None,
402 ) -> Result[T, PermissionDenied | AdminValidationError | NotFoundError]:
403 """Update an existing resource.
405 Args:
406 item_id: Resource identifier
407 data: Update data
408 user: Current user for authorization
410 Returns:
411 Result containing updated resource on success, error on failure
412 """
413 # Check if record exists
414 record = await self.data_source.find_one(item_id)
415 if record is None:
416 return Err(
417 NotFoundError(
418 resource=self.resource_name,
419 identifier=str(item_id),
420 message=f"{self.resource_name} not found",
421 ),
422 )
424 # Check authorization
425 if not await self.authorizer.can_update(user, self.resource_name, record):
426 return Err(
427 PermissionDenied(
428 resource=self.resource_name,
429 action="update",
430 message=f"Cannot update this {self.resource_name}",
431 ),
432 )
434 # Validate data
435 validation_result = await self.validator.validate(data)
436 if validation_result.is_err():
437 return validation_result # type: ignore[return-value]
439 validated_data = validation_result.unwrap()
440 updated = await self.data_source.update(item_id, validated_data)
442 if self.model and isinstance(updated, dict):
443 updated = self.model(**updated)
445 # Record successful update audit event
446 resource_id = getattr(updated, "id", str(item_id))
447 await self._record_audit(
448 action="admin.resource.update",
449 actor=user,
450 resource_id=resource_id,
451 outcome="success",
452 severity=AuditEventSeverity.MEDIUM,
453 )
455 return Ok(updated)
457 async def _delete_internal(
458 self,
459 item_id: Any,
460 user: Any = None,
461 *,
462 skip_audit: bool = False,
463 ) -> Result[bool, PermissionDenied | NotFoundError]:
464 """Internal delete implementation shared by delete() and bulk_delete().
466 Args:
467 item_id: Resource identifier
468 user: Current user for authorization
469 skip_audit: If True, skip audit recording (used by bulk_delete fallback)
471 Returns:
472 Result containing True on success, error on failure
473 """
474 # Check if record exists
475 record = await self.data_source.find_one(item_id)
476 if record is None:
477 return Err(
478 NotFoundError(
479 resource=self.resource_name,
480 identifier=str(item_id),
481 message=f"{self.resource_name} not found",
482 ),
483 )
485 # Check authorization
486 if not await self.authorizer.can_delete(user, self.resource_name, record):
487 return Err(
488 PermissionDenied(
489 resource=self.resource_name,
490 action="delete",
491 message=f"Cannot delete this {self.resource_name}",
492 ),
493 )
495 await self.data_source.delete(item_id)
497 # Record successful deletion audit event unless skipped (bulk_delete handles its own audit)
498 if not skip_audit:
499 await self._record_audit(
500 action="admin.resource.delete",
501 actor=user,
502 resource_id=str(item_id),
503 outcome="success",
504 severity=AuditEventSeverity.CRITICAL,
505 )
507 return Ok(True)
509 async def delete(
510 self,
511 item_id: Any,
512 *,
513 user: Any = None,
514 ) -> Result[bool, PermissionDenied | NotFoundError]:
515 """Delete a resource.
517 Args:
518 item_id: Resource identifier
519 user: Current user for authorization
521 Returns:
522 Result containing True on success, error on failure
523 """
524 return await self._delete_internal(item_id, user)
526 async def bulk_delete(
527 self,
528 ids: list[Any], # type: ignore[valid-type]
529 *,
530 user: Any = None,
531 ) -> Result[int, PermissionDenied]:
532 """Delete multiple resources.
534 Args:
535 ids: List of resource identifiers
536 user: Current user for authorization
538 Returns:
539 Result containing count of deleted resources, error on failure
540 """
541 # Check bulk authorization
542 if not await self.authorizer.can_delete(user, self.resource_name):
543 return Err(
544 PermissionDenied(
545 resource=self.resource_name,
546 action="delete",
547 message=f"Cannot delete {self.resource_name}",
548 ),
549 )
551 # Check ResourceDataSourceProtocol supports bulk operations
552 if hasattr(self.data_source, "delete_many"):
553 count = await self.data_source.delete_many(ids)
554 else:
555 # Fallback to individual deletes without per-item audit events
556 count = 0
557 if self._uow is not None:
558 async with self._uow:
559 for id_ in ids: # type: ignore[attr-defined]
560 result = await self._delete_internal(
561 id_, user=user, skip_audit=True
562 )
563 if result.is_ok():
564 count += 1
565 else:
566 for id_ in ids: # type: ignore[attr-defined]
567 result = await self._delete_internal(
568 id_, user=user, skip_audit=True
569 )
570 if result.is_ok():
571 count += 1
573 # Record single bulk deletion audit event for both paths
574 await self._record_audit(
575 action="admin.resource.bulk_delete",
576 actor=user,
577 resource_id="bulk",
578 outcome="success",
579 severity=AuditEventSeverity.CRITICAL,
580 deleted_count=count,
581 requested_ids=len(ids),
582 )
584 return Ok(count)
586 async def bulk_update(
587 self,
588 updates: list[tuple[Any, dict[str, Any]]], # type: ignore[valid-type]
589 *,
590 user: Any = None,
591 ) -> Result[int, PermissionDenied]:
592 """Update multiple resources.
594 Args:
595 updates: List of (id, data) tuples
596 user: Current user for authorization
598 Returns:
599 Result containing count of updated resources, error on failure
600 """
601 # Check bulk authorization
602 if not await self.authorizer.can_update(user, self.resource_name):
603 return Err(
604 PermissionDenied(
605 resource=self.resource_name,
606 action="update",
607 message=f"Cannot update {self.resource_name}",
608 ),
609 )
611 # Check ResourceDataSourceProtocol supports bulk operations
612 if hasattr(self.data_source, "update_many"):
613 count = await self.data_source.update_many(updates)
614 return Ok(count)
616 # Fallback to individual updates
617 count = 0
618 if self._uow is not None:
619 async with self._uow:
620 for id_, data in updates: # type: ignore[attr-defined]
621 result = await self.update(id_, data, user=user)
622 if result.is_ok():
623 count += 1
624 else:
625 for id_, data in updates: # type: ignore[attr-defined]
626 result = await self.update(id_, data, user=user)
627 if result.is_ok():
628 count += 1
630 return Ok(count)