Coverage for src / lexigram / contracts / data / repository.py: 0%

1"""RepositoryProtocol protocols. 

2 

3Generic repository patterns for data access abstraction. 

4""" 

5 

6from __future__ import annotations 

7 

8from typing import TYPE_CHECKING, Any, Generic, Protocol, TypeVar, runtime_checkable 

9 

10if TYPE_CHECKING: 

11 from lexigram.contracts.domain.specification import SpecificationProtocol 

12 

13T = TypeVar("T") 

14 

15 

16@runtime_checkable 

17class ReadOnlyRepositoryProtocol(Protocol, Generic[T]): 

18 """Protocol for read-only repository operations. 

19 

20 Use this for query-only access patterns (CQRS query side). 

21 

22 Example: 

23 ```python 

24 class UserQueryRepository: 

25 async def get(self, id: str) -> User | None: 

26 return await self.db.query("users").where(id=id).first() 

27 

28 async def list(self, skip: int = 0, limit: int = 100) -> list[User]: 

29 return await self.db.query("users").offset(skip).limit(limit).all() 

30 ``` 

31 """ 

32 

33 async def get(self, item_id: str) -> T | None: 

34 """Get entity by ID. 

35 

36 Args: 

37 item_id: Entity identifier. 

38 

39 Returns: 

40 Entity if found, None otherwise. 

41 """ 

42 ... 

43 

44 async def list( 

45 self, 

46 skip: int = 0, 

47 limit: int = 100, 

48 **filters: Any, 

49 ) -> list[T]: 

50 """List entities with pagination. 

51 

52 Args: 

53 skip: Number of records to skip. 

54 limit: Maximum records to return. 

55 **filters: Optional filter criteria. 

56 

57 Returns: 

58 List of entities. 

59 """ 

60 ... 

61 

62 async def find_by_spec(self, spec: SpecificationProtocol[T]) -> list[T]: # type: ignore[valid-type] 

63 """Find entities matching a complex specification. 

64 

65 Args: 

66 spec: The DDD specification to evaluate. 

67 

68 Returns: 

69 List of matching entities. 

70 """ 

71 ... 

72 

73 async def count(self, **filters: Any) -> int: 

74 """Count entities matching filters. 

75 

76 Args: 

77 **filters: Optional filter criteria. 

78 

79 Returns: 

80 Total count of matching entities. 

81 """ 

82 ... 

83 

84 

85@runtime_checkable 

86class RepositoryProtocol(ReadOnlyRepositoryProtocol[T], Protocol, Generic[T]): 

87 """Protocol for full repository operations. 

88 

89 Extends ReadOnlyRepositoryProtocol with write operations. 

90 

91 Example: 

92 ```python 

93 class UserRepository: 

94 async def save(self, entity: User) -> User: 

95 if entity.id: 

96 await self.db.update("users", entity.dict()).where(id=entity.id) 

97 else: 

98 entity.id = await self.db.insert("users", entity.dict()) 

99 return entity 

100 

101 async def delete(self, id: str) -> bool: 

102 result = await self.db.delete("users").where(id=id) 

103 return result.affected_rows > 0 

104 ``` 

105 """ 

106 

107 async def save(self, entity: T) -> T: 

108 """Save (create or update) an entity. 

109 

110 Args: 

111 entity: Entity to save. 

112 

113 Returns: 

114 Saved entity with any generated fields populated. 

115 """ 

116 ... 

117 

118 async def delete(self, item_id: str) -> bool: 

119 """Delete entity by ID. 

120 

121 Args: 

122 item_id: Entity identifier. 

123 

124 Returns: 

125 True if deleted, False if not found. 

126 """ 

127 ... 

128 

129 async def save_many(self, entities: list[T]) -> list[T]: 

130 """Save (create or update) multiple entities in a single operation. 

131 

132 Implementations SHOULD execute this as a batch for efficiency. 

133 

134 Args: 

135 entities: Entities to save. 

136 

137 Returns: 

138 Saved entities with any generated fields populated. 

139 """ 

140 ... 

141 

142 async def delete_many(self, item_ids: list[str]) -> int: 

143 """Delete multiple entities by ID. 

144 

145 Implementations SHOULD execute this as a batch for efficiency. 

146 

147 Args: 

148 item_ids: Entity identifiers to delete. 

149 

150 Returns: 

151 Number of entities actually deleted. 

152 """ 

153 ... 

154 

155 

156__all__ = [ 

157 "ReadOnlyRepositoryProtocol", 

158 "RepositoryProtocol", 

159 "T", 

160]