Coverage for src/lexigram/auth/storage/token_store.py: 88%

1"""User and token storage interfaces and implementations""" 

2 

3from __future__ import annotations 

4 

5import secrets 

6from typing import Any, Protocol, runtime_checkable 

7 

8from lexigram.auth.models.user import User, UserCredentials 

9 

10 

11@runtime_checkable 

12class CachedUserStore(Protocol): 

13 """Protocol for read-through cache user stores (point-lookup only). 

14 

15 Implementations such as :class:`RedisUserStore` are highly efficient for 

16 ``get_user_by_id`` but are **not** designed for full enumeration 

17 operations like ``list_users`` or ``count_users``. Use a relational or 

18 document-oriented :class:`UserStoreProtocol` as the primary source of truth and 

19 pair it with a :class:`CachedUserStore` for hot-path reads. 

20 """ 

21 

22 async def get_user_by_id(self, user_id: str) -> User | None: 

23 """Return the user with *user_id*, or ``None`` if not found.""" 

24 ... 

25 

26 async def create_user( 

27 self, 

28 name: str, 

29 email: str, 

30 hashed_password: str | None, 

31 roles: list[str] | None = None, 

32 permissions: list[str] | None = None, 

33 profile: dict[str, Any] | None = None, 

34 **kwargs: Any, 

35 ) -> User: 

36 """Create and cache a user entry.""" 

37 ... 

38 

39 async def update_user(self, user: User) -> None: 

40 """Update a cached user entry.""" 

41 ... 

42 

43 async def delete_user(self, user_id: str) -> None: 

44 """Evict and delete a cached user entry.""" 

45 ... 

46 

47 

48@runtime_checkable 

49class UserStoreProtocol(Protocol): 

50 """Protocol for user storage implementations.""" 

51 

52 async def create_user( 

53 self, 

54 name: str, 

55 email: str, 

56 hashed_password: str | None, 

57 roles: list[str] | None = None, 

58 permissions: list[str] | None = None, 

59 profile: dict[str, Any] | None = None, 

60 **kwargs: Any, 

61 ) -> User: 

62 """Create a new user. 

63 

64 Accepts arbitrary keyword arguments for backwards compatibility 

65 (e.g. ``is_verified``) which may be ignored by some implementations. 

66 Stores credential data (``hashed_password``) internally and makes it 

67 accessible via :meth:`get_credentials`. 

68 """ 

69 ... 

70 

71 async def get_user_by_id(self, user_id: str) -> User | None: 

72 """Get user by ID""" 

73 ... 

74 

75 async def get_user_by_email(self, email: str) -> User | None: 

76 """Get user by email""" 

77 ... 

78 

79 async def update_user(self, user: User) -> None: 

80 """Update non-credential user information. 

81 

82 Does **not** update password hashes. Use :meth:`update_credentials` 

83 for password changes. 

84 """ 

85 ... 

86 

87 async def delete_user(self, user_id: str) -> None: 

88 """Delete a user""" 

89 ... 

90 

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

92 """List users with pagination""" 

93 ... 

94 

95 async def count_users(self) -> int: 

96 """Count total users""" 

97 ... 

98 

99 async def get_credentials(self, user_id: str) -> UserCredentials | None: 

100 """Return the :class:`UserCredentials` for *user_id*, or ``None``. 

101 

102 Use this for all authentication and password-related operations 

103 instead of reading fields directly from :class:`User`. 

104 """ 

105 ... 

106 

107 async def update_credentials(self, creds: UserCredentials) -> None: 

108 """Persist updated credential data for the user identified by 

109 ``creds.user_id``. 

110 

111 This is the only sanctioned way to change a stored password hash. 

112 """ 

113 ... 

114 

115 

116class InMemoryUserStore(UserStoreProtocol): 

117 """Simple in-memory user store for development/testing""" 

118 

119 def __init__(self) -> None: 

120 self.users: dict[str, User] = {} 

121 self.name_index: dict[str, str] = {} 

122 self.email_index: dict[str, str] = {} 

123 # Credentials stored separately from public User data 

124 self._credentials: dict[str, UserCredentials] = {} 

125 

126 async def create_user( 

127 self, 

128 name: str, 

129 email: str, 

130 hashed_password: str | None, 

131 roles: list[str] | None = None, 

132 permissions: list[str] | None = None, 

133 profile: dict[str, Any] | None = None, 

134 **kwargs: Any, 

135 ) -> User: 

136 """Create a new user""" 

137 if name in self.name_index: 

138 raise ValueError(f"Name '{name}' already exists") 

139 if email in self.email_index: 

140 raise ValueError(f"Email '{email}' already exists") 

141 

142 user_id = secrets.token_urlsafe(16) 

143 

144 is_verified = bool(kwargs.get("is_verified", False)) 

145 

146 user = User( 

147 user_id=user_id, 

148 name=name, 

149 email=email, 

150 roles=list(roles or []), 

151 permissions=list(permissions or []), 

152 profile=profile or {}, 

153 is_verified=is_verified, 

154 ) 

155 

156 self.users[user_id] = user 

157 self.name_index[name] = user_id 

158 self.email_index[email] = user_id 

159 # Store credentials separately 

160 self._credentials[user_id] = UserCredentials( 

161 user_id=user_id, 

162 hashed_password=hashed_password, 

163 ) 

164 

165 return user 

166 

167 async def get_user_by_id(self, user_id: str) -> User | None: 

168 """Get user by ID""" 

169 return self.users.get(user_id) 

170 

171 async def get_user_by_email(self, email: str) -> User | None: 

172 """Get user by email""" 

173 user_id = self.email_index.get(email) 

174 return self.users.get(user_id) if user_id else None 

175 

176 async def get_user_by_username(self, username: str) -> User | None: 

177 """Get user by username (name).""" 

178 user_id = self.name_index.get(username) 

179 return self.users.get(user_id) if user_id else None 

180 

181 async def update_user(self, user: User) -> None: 

182 """Update user information""" 

183 if user.user_id not in self.users: 

184 raise ValueError(f"User '{user.user_id}' not found") 

185 

186 self.users[user.user_id] = user 

187 # Update indexes if name or email changed 

188 if user.name is not None: 

189 self.name_index[user.name] = user.user_id 

190 if user.email is not None: 

191 self.email_index[user.email] = user.user_id 

192 

193 async def delete_user(self, user_id: str) -> None: 

194 """Delete a user""" 

195 user = self.users.get(user_id) 

196 if user: 

197 del self.users[user_id] 

198 if user.name is not None: 

199 del self.name_index[user.name] 

200 if user.email is not None: 

201 del self.email_index[user.email] 

202 self._credentials.pop(user_id, None) 

203 

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

205 """List users with pagination""" 

206 all_users = list(self.users.values()) 

207 return all_users[skip : skip + limit] 

208 

209 async def count_users(self) -> int: 

210 """Count total users""" 

211 return len(self.users) 

212 

213 async def get_credentials(self, user_id: str) -> UserCredentials | None: 

214 """Return stored credentials for *user_id*.""" 

215 return self._credentials.get(user_id) 

216 

217 async def update_credentials(self, creds: UserCredentials) -> None: 

218 """Persist updated credentials for the user.""" 

219 if creds.user_id not in self.users: 

220 raise ValueError(f"User '{creds.user_id}' not found") 

221 self._credentials[creds.user_id] = creds 

222 

223 

224__all__ = ["CachedUserStore", "InMemoryUserStore", "UserStoreProtocol"]