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

1"""MongoDB-backed user store implementation. 

2 

3Migrated to use ``DocumentStoreProtocol`` from ``lexigram-nosql`` for 

4connection lifecycle and collection access. 

5""" 

6 

7from __future__ import annotations 

8 

9from typing import TYPE_CHECKING, Any 

10 

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

12from lexigram.di.decorators import inject 

13from lexigram.logging import get_logger 

14 

15if TYPE_CHECKING: 

16 from lexigram.contracts.data import CollectionProtocol, DocumentStoreProtocol 

17 

18logger = get_logger(__name__) 

19 

20 

21@inject 

22class MongoDBUserStore: 

23 """MongoDB-based user store using ``DocumentStoreProtocol``.""" 

24 

25 def __init__( 

26 self, 

27 document_store: DocumentStoreProtocol, 

28 collection_name: str = "users", 

29 ): 

30 self._store = document_store 

31 self.collection_name = collection_name 

32 self._collection: CollectionProtocol | None = None 

33 self._initialized = False 

34 

35 @property 

36 def _col(self) -> CollectionProtocol: 

37 """Lazily initialize the collection handle.""" 

38 if self._collection is None: 

39 self._collection = self._store.collection(self.collection_name) 

40 return self._collection 

41 

42 async def _ensure_collection(self) -> None: 

43 """Ensure user collection exists.""" 

44 if self._initialized: 

45 return 

46 # MongoDB collections are created automatically on first insert 

47 self._initialized = True 

48 

49 async def _user_from_doc(self, doc: dict[str, Any]) -> User: 

50 """Convert MongoDB document to User object.""" 

51 return User( 

52 user_id=doc["_id"], 

53 name=doc["name"] if "name" in doc else doc.get("username"), 

54 email=doc["email"], 

55 is_active=doc.get("is_active", True), 

56 is_verified=doc.get("is_verified", False), 

57 roles=doc.get("roles", []), 

58 permissions=doc.get("permissions", []), 

59 profile=doc.get("profile", {}), 

60 created_at=doc.get("created_at"), 

61 updated_at=doc.get("updated_at"), 

62 last_login_at=doc.get("last_login_at"), 

63 login_count=doc.get("login_count", 0), 

64 ) 

65 

66 async def _doc_from_user(self, user: User) -> dict[str, Any]: 

67 """Convert User object to MongoDB document (non-credential fields).""" 

68 return { 

69 "_id": user.user_id, 

70 "name": user.name, 

71 "email": user.email, 

72 "is_active": user.is_active, 

73 "is_verified": user.is_verified, 

74 "roles": user.roles, 

75 "permissions": user.permissions, 

76 "profile": user.profile, 

77 "created_at": user.created_at, 

78 "updated_at": user.updated_at, 

79 "last_login_at": user.last_login_at, 

80 "login_count": user.login_count, 

81 } 

82 

83 async def create_user( 

84 self, 

85 name: str, 

86 email: str, 

87 hashed_password: str | None, 

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

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

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

91 **kwargs: Any, 

92 ) -> User: 

93 """Create a new user.""" 

94 await self._ensure_collection() 

95 

96 import uuid 

97 

98 user_id = str(uuid.uuid4()) 

99 

100 user = User( 

101 user_id=user_id, 

102 name=name, 

103 email=email, 

104 roles=list(roles or []), 

105 permissions=list(permissions or []), 

106 profile=profile or {}, 

107 ) 

108 

109 doc = await self._doc_from_user(user) 

110 doc["hashed_password"] = hashed_password 

111 doc["previous_passwords"] = [] 

112 

113 await self._col.insert_one(doc) 

114 

115 logger.info("Created user: %s", name) 

116 return user 

117 

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

119 """Get user by ID.""" 

120 await self._ensure_collection() 

121 

122 doc = await self._col.find_one({"_id": user_id}) 

123 return await self._user_from_doc(doc) if doc else None 

124 

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

126 """Get user by email.""" 

127 await self._ensure_collection() 

128 

129 doc = await self._col.find_one({"email": email}) 

130 return await self._user_from_doc(doc) if doc else None 

131 

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

133 """Update non-credential user information.""" 

134 await self._ensure_collection() 

135 

136 doc = await self._doc_from_user(user) 

137 await self._col.replace_one({"_id": user.user_id}, doc, upsert=True) 

138 

139 logger.info("Updated user: %s", user.name) 

140 

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

142 """Delete a user.""" 

143 await self._ensure_collection() 

144 

145 await self._col.delete_one({"_id": user_id}) 

146 logger.info("Deleted user: %s", user_id) 

147 

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

149 """List users with pagination.""" 

150 await self._ensure_collection() 

151 

152 users = [] 

153 cursor = self._col.find({}, skip=skip, limit=limit) 

154 async for doc in cursor: # type: ignore[attr-defined] 

155 user = await self._user_from_doc(doc) 

156 users.append(user) 

157 

158 return users 

159 

160 async def count_users(self) -> int: 

161 """Count total users.""" 

162 await self._ensure_collection() 

163 

164 return await self._col.count_documents({}) 

165 

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

167 """Return credential data for *user_id*.""" 

168 await self._ensure_collection() 

169 doc = await self._col.find_one( 

170 {"_id": user_id}, 

171 projection={"hashed_password": 1, "previous_passwords": 1}, 

172 ) 

173 if not doc: 

174 return None 

175 return UserCredentials( 

176 user_id=user_id, 

177 hashed_password=doc.get("hashed_password"), 

178 previous_hashes=doc.get("previous_passwords", []), 

179 ) 

180 

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

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

183 await self._ensure_collection() 

184 await self._col.update_one( 

185 {"_id": creds.user_id}, 

186 { 

187 "$set": { 

188 "hashed_password": creds.hashed_password, 

189 "previous_passwords": creds.previous_hashes, 

190 } 

191 }, 

192 ) 

193 logger.info("Updated credentials for user: %s", creds.user_id) 

194 

195 

196__all__ = ["MongoDBUserStore"]