Coverage for src/lexigram/auth/storage/db_stores.py: 26%
95 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 12:26 +0800
1"""Database-backed user store implementations."""
3from __future__ import annotations
5from typing import TYPE_CHECKING, Any
7from lexigram.auth.models.user import User, UserCredentials
8from lexigram.auth.storage._mongo_store import MongoDBUserStore
9from lexigram.auth.storage._sql_store import SQLUserStore
10from lexigram.di.decorators import inject
11from lexigram.logging import get_logger
12from lexigram.serialization import dumps, loads
14if TYPE_CHECKING:
15 from lexigram.contracts import DatabaseProviderProtocol
17logger = get_logger(__name__)
20@inject
21class RedisUserStore:
22 """Redis-backed user cache (read-through cache, not primary storage).
24 Implements :class:`~lexigram.auth.storage.token_store.CachedUserStore` —
25 it is designed for **point-lookup** operations only. Full-scan operations
26 (``list_users``, ``count_users``, ``get_user_by_email``) are intentionally
27 not supported; a relational or document-oriented
28 :class:`~lexigram.auth.storage.token_store.UserStoreProtocol` must be used as the
29 primary source of truth.
31 Stores user records as Redis hashes under ``{prefix}{user_id}``.
32 Credential fields (``hashed_password``, ``previous_passwords``) are kept
33 in the same Redis hash but are only accessible via :meth:`get_credentials`
34 and mutated via :meth:`update_credentials`.
36 Args:
37 db_provider: Database provider whose ``redis`` attribute exposes
38 a Redis-compatible async client.
39 prefix: Key prefix applied to every user key. Defaults to
40 ``"user:"``.
41 ttl: Optional time-to-live in seconds applied to every cached
42 user entry. ``None`` means entries never expire (default).
44 .. warning::
45 Do **not** use :class:`RedisUserStore` for operations that require
46 full enumeration (``list_users``, ``count_users``,
47 ``get_user_by_email``). These methods raise :exc:`NotImplementedError`
48 by design. Use a SQL or MongoDB store for those operations.
49 """
51 def __init__(
52 self,
53 db_provider: DatabaseProviderProtocol,
54 prefix: str = "user:",
55 ttl: int | None = None,
56 ) -> None:
57 self.db_provider = db_provider
58 self.prefix = prefix
59 self.ttl = ttl
61 async def _user_key(self, user_id: str) -> str:
62 """Generate Redis key for user"""
63 return f"{self.prefix}{user_id}"
65 async def _user_from_data(self, data: dict[str, Any]) -> User:
66 """Convert Redis hash data to a User object (credentials excluded)."""
67 from datetime import datetime
69 def parse_dt(val: Any) -> datetime | None:
70 if not val:
71 return None
72 if isinstance(val, datetime):
73 return val
74 try:
75 if isinstance(val, (int, float)):
76 return datetime.fromtimestamp(val)
77 return datetime.fromisoformat(str(val))
78 except (ValueError, TypeError):
79 return None
81 return User(
82 user_id=data["id"],
83 name=data["name"] if "name" in data else data.get("username"),
84 email=data["email"],
85 is_active=data.get("is_active", True),
86 is_verified=data.get("is_verified", False),
87 roles=loads(data.get("roles", "[]")),
88 permissions=loads(data.get("permissions", "[]")),
89 profile=loads(data.get("profile", "{}")),
90 created_at=parse_dt(data.get("created_at")),
91 updated_at=parse_dt(data.get("updated_at")),
92 last_login_at=parse_dt(data.get("last_login_at")),
93 login_count=int(data.get("login_count", 0)),
94 )
96 async def _data_from_user(self, user: User) -> dict[str, str]:
97 """Convert User object to Redis hash fields (non-credential fields only)."""
98 data: dict[str, str] = {
99 "id": user.user_id,
100 "name": user.name or "",
101 "email": user.email,
102 "is_active": str(user.is_active),
103 "is_verified": str(user.is_verified),
104 "roles": str(dumps(user.roles)),
105 "permissions": str(dumps(user.permissions)),
106 "profile": str(dumps(user.profile)),
107 "login_count": str(user.login_count),
108 }
110 if user.created_at:
111 data["created_at"] = user.created_at.isoformat()
112 if user.updated_at:
113 data["updated_at"] = user.updated_at.isoformat()
114 if user.last_login_at:
115 data["last_login_at"] = user.last_login_at.isoformat()
117 return data
119 async def create_user(
120 self,
121 name: str,
122 email: str,
123 hashed_password: str | None,
124 roles: list[str] | None = None,
125 permissions: list[str] | None = None,
126 profile: dict[str, Any] | None = None,
127 **kwargs: Any,
128 ) -> User:
129 """Cache a new user entry.
131 .. note::
132 Redis is typically used as a read-through cache. For persistent
133 primary storage use :class:`SQLUserStore` or :class:`MongoDBUserStore`.
134 """
135 import uuid
137 user_id = str(uuid.uuid4())
139 user = User(
140 user_id=user_id,
141 name=name,
142 email=email,
143 roles=list(roles or []),
144 permissions=list(permissions or []),
145 profile=profile or {},
146 )
148 key = await self._user_key(user_id)
149 data = await self._data_from_user(user)
150 # Store credential fields alongside other fields in the same hash
151 if hashed_password:
152 data["hashed_password"] = hashed_password
153 data["previous_passwords"] = str(dumps([]))
155 await self.db_provider.redis.hset(key, mapping=data) # type: ignore[attr-defined]
156 if self.ttl is not None:
157 await self.db_provider.redis.expire(key, self.ttl) # type: ignore[attr-defined]
159 logger.info("Cached user: %s", name)
160 return user
162 async def get_user_by_id(self, user_id: str) -> User | None:
163 """Get user by ID"""
164 key = await self._user_key(user_id)
165 data = await self.db_provider.redis.hgetall(key) # type: ignore[attr-defined]
167 if not data:
168 return None
170 str_data = {
171 k.decode() if isinstance(k, bytes) else k: (
172 v.decode() if isinstance(v, bytes) else v
173 )
174 for k, v in data.items()
175 }
177 return await self._user_from_data(str_data)
179 async def get_user_by_email(self, email: str) -> User | None:
180 """Not supported — Redis does not index by email efficiently.
182 .. warning::
183 This always raises :exc:`NotImplementedError`. Use the primary
184 relational/document store for email-based lookups.
185 """
186 raise NotImplementedError(
187 "RedisUserStore (CachedUserStore) does not support email lookups. "
188 "Use the primary UserStoreProtocol for get_user_by_email.",
189 )
191 async def update_user(self, user: User) -> None:
192 """Update non-credential fields for the cached user entry."""
193 key = await self._user_key(user.user_id)
194 data = await self._data_from_user(user)
196 await self.db_provider.redis.hset(key, mapping=data) # type: ignore[attr-defined]
197 if self.ttl is not None:
198 await self.db_provider.redis.expire(key, self.ttl) # type: ignore[attr-defined]
200 logger.info("Updated cached user: %s", user.name)
202 async def delete_user(self, user_id: str) -> None:
203 """Delete a user"""
204 key = await self._user_key(user_id)
205 await self.db_provider.redis.delete(key) # type: ignore[attr-defined]
207 logger.info("Deleted cached user: %s", user_id)
209 async def list_users(self, skip: int = 0, limit: int = 100) -> list[User]:
210 """Not supported — Redis is not designed for full user enumeration.
212 .. warning::
213 Always raises :exc:`NotImplementedError`. Use the primary
214 relational/document store for listing users.
215 """
216 raise NotImplementedError(
217 "RedisUserStore (CachedUserStore) does not support list_users. "
218 "Use the primary UserStoreProtocol for enumeration operations.",
219 )
221 async def count_users(self) -> int:
222 """Not supported — Redis is not designed for counting all users.
224 .. warning::
225 Always raises :exc:`NotImplementedError`. Use the primary
226 relational/document store for count operations.
227 """
228 raise NotImplementedError(
229 "RedisUserStore (CachedUserStore) does not support count_users. "
230 "Use the primary UserStoreProtocol for enumeration operations.",
231 )
233 async def get_credentials(self, user_id: str) -> UserCredentials | None:
234 """Return cached credential data for *user_id*.
236 Returns ``None`` if the user entry is not in the cache.
237 """
238 key = await self._user_key(user_id)
239 data = await self.db_provider.redis.hgetall(key) # type: ignore[attr-defined]
240 if not data:
241 return None
242 str_data = {
243 k.decode() if isinstance(k, bytes) else k: (
244 v.decode() if isinstance(v, bytes) else v
245 )
246 for k, v in data.items()
247 }
248 return UserCredentials(
249 user_id=str_data.get("id", user_id),
250 hashed_password=str_data.get("hashed_password"),
251 previous_hashes=loads(str_data.get("previous_passwords", "[]")),
252 )
254 async def update_credentials(self, creds: UserCredentials) -> None:
255 """Update credential fields in the cached user hash."""
256 key = await self._user_key(creds.user_id)
257 data: dict[str, str] = {
258 "previous_passwords": str(dumps(creds.previous_hashes)),
259 }
260 if creds.hashed_password is not None:
261 data["hashed_password"] = creds.hashed_password
263 await self.db_provider.redis.hset(key, mapping=data) # type: ignore[attr-defined]
264 if self.ttl is not None:
265 await self.db_provider.redis.expire(key, self.ttl) # type: ignore[attr-defined]
267 logger.info("Updated cached credentials for user: %s", creds.user_id)
270__all__ = ["MongoDBUserStore", "RedisUserStore", "SQLUserStore"]