Coverage for src / lexigram / contracts / auth / store.py: 100%

1"""User store protocols. 

2 

3Segregated interfaces for user data access following ISP. 

4""" 

5 

6from __future__ import annotations 

7 

8from typing import TYPE_CHECKING, Protocol, runtime_checkable 

9 

10if TYPE_CHECKING: 

11 from lexigram.contracts.auth.user import UserProtocol 

12 

13 

14@runtime_checkable 

15class UserReaderProtocol(Protocol): 

16 """Protocol for reading user data. 

17 

18 Use for query-only operations (CQRS query side). 

19 """ 

20 

21 async def get_by_id(self, user_id: str) -> UserProtocol | None: 

22 """Get user by ID. 

23 

24 Args: 

25 user_id: User identifier. 

26 

27 Returns: 

28 User if found, None otherwise. 

29 """ 

30 ... 

31 

32 async def get_by_email(self, email: str) -> UserProtocol | None: 

33 """Get user by email. 

34 

35 Args: 

36 email: User email address. 

37 

38 Returns: 

39 User if found, None otherwise. 

40 """ 

41 ... 

42 

43 async def get_by_username(self, username: str) -> UserProtocol | None: 

44 """Get user by username. 

45 

46 Args: 

47 username: Username to look up. 

48 

49 Returns: 

50 User if found, None otherwise. 

51 """ 

52 ... 

53 

54 async def list_users( 

55 self, 

56 skip: int = 0, 

57 limit: int = 100, 

58 ) -> list[UserProtocol]: 

59 """List users with pagination. 

60 

61 Args: 

62 skip: Number of records to skip. 

63 limit: Maximum records to return. 

64 

65 Returns: 

66 List of users. 

67 """ 

68 ... 

69 

70 async def count_users(self) -> int: 

71 """Count total users. 

72 

73 Returns: 

74 Total user count. 

75 """ 

76 ... 

77 

78 

79@runtime_checkable 

80class UserWriterProtocol(Protocol): 

81 """Protocol for writing user data. 

82 

83 Use for command-only operations (CQRS command side). 

84 """ 

85 

86 async def create(self, user: UserProtocol) -> UserProtocol: 

87 """Create a new user. 

88 

89 Args: 

90 user: User to create. 

91 

92 Returns: 

93 Created user. 

94 """ 

95 ... 

96 

97 async def update(self, user: UserProtocol) -> UserProtocol: 

98 """Update an existing user. 

99 

100 Args: 

101 user: User to update. 

102 

103 Returns: 

104 Updated user. 

105 """ 

106 ... 

107 

108 async def delete(self, user_id: str) -> bool: 

109 """Delete a user. 

110 

111 Args: 

112 user_id: User identifier. 

113 

114 Returns: 

115 True if deleted, False if not found. 

116 """ 

117 ... 

118 

119 

120@runtime_checkable 

121class UserStoreProtocol(UserReaderProtocol, UserWriterProtocol, Protocol): 

122 """Combined user store protocol for read/write operations. 

123 

124 This protocol combines UserReaderProtocol and UserWriterProtocol for convenience 

125 in implementations that provide both read and write capabilities. 

126 """ 

127 

128 

129__all__ = [ 

130 "UserReaderProtocol", 

131 "UserStoreProtocol", 

132 "UserWriterProtocol", 

133]