Coverage for src/lexigram/admin/data/admin_repository.py: 97%

1"""In-memory repository implementation for AdminUserAggregate. 

2 

3Provides a concrete ``AbstractRepository`` implementation backed by an 

4in-memory dict, suitable for testing and lightweight scenarios. 

5""" 

6 

7from __future__ import annotations 

8 

9from typing import TYPE_CHECKING, Any 

10 

11from lexigram.admin.domain.aggregate import AdminUserAggregate 

12from lexigram.primitives.data import AbstractRepository 

13 

14if TYPE_CHECKING: 

15 from lexigram.contracts.domain.specification import SpecificationProtocol 

16 

17 

18class AdminUserRepository(AbstractRepository[AdminUserAggregate, str]): 

19 """In-memory repository for AdminUserAggregate instances. 

20 

21 Stores aggregates in a plain dict keyed by their ``id``. Intended for 

22 use in tests and as a reference implementation; production deployments 

23 should supply a database-backed subclass. 

24 """ 

25 

26 def __init__(self) -> None: 

27 super().__init__() 

28 self._store: dict[str, AdminUserAggregate] = {} 

29 

30 # ------------------------------------------------------------------ 

31 # Read primitives 

32 # ------------------------------------------------------------------ 

33 

34 async def _fetch_by_id(self, entity_id: Any) -> AdminUserAggregate | None: 

35 return self._store.get(str(entity_id)) 

36 

37 async def _fetch_many( 

38 self, 

39 *, 

40 skip: int, 

41 limit: int, 

42 filters: dict[str, Any], 

43 ) -> list[AdminUserAggregate]: 

44 items = list(self._store.values()) 

45 for field_name, value in filters.items(): 

46 items = [i for i in items if getattr(i, field_name, None) == value] 

47 return items[skip : skip + limit] 

48 

49 async def _count(self, *, filters: dict[str, Any]) -> int: 

50 items = list(self._store.values()) 

51 for field_name, value in filters.items(): 

52 items = [i for i in items if getattr(i, field_name, None) == value] 

53 return len(items) 

54 

55 async def find_by_spec( 

56 self, 

57 spec: SpecificationProtocol[AdminUserAggregate], 

58 ) -> list[AdminUserAggregate]: 

59 """Return all aggregates that satisfy *spec*.""" 

60 return [item for item in self._store.values() if spec.is_satisfied_by(item)] 

61 

62 # ------------------------------------------------------------------ 

63 # Write primitives 

64 # ------------------------------------------------------------------ 

65 

66 async def _save(self, entity: AdminUserAggregate) -> AdminUserAggregate: 

67 self._store[str(entity.id)] = entity 

68 return entity 

69 

70 async def _delete(self, entity_id: Any) -> bool: 

71 key = str(entity_id) 

72 if key not in self._store: 

73 return False 

74 del self._store[key] 

75 return True