Coverage for src / lexigram / admin / auth / store / direct_sql.py: 9%
223 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-13 22:14 +0800
1"""
2Direct SQL admin user store implementation.
3"""
5from __future__ import annotations
7from typing import Any
8import uuid
10from lexigram.contracts.data import DatabaseProviderProtocol
11from lexigram.di.decorators import inject
12from lexigram.logging import get_logger
14logger = get_logger(__name__)
17@inject
18class DirectSQLAdminUserStore:
19 """Simple store implementation for admin_users table using direct SQL.
21 This class provides the subset of operations required by AdminAuthAdapter
22 (create_user, get_user_by_username, update_user, delete_user).
23 Renamed from AdminUserStore to avoid conflict with config-backed store.
24 """
26 def __init__(self, db_provider: DatabaseProviderProtocol) -> None:
27 self.db_provider = db_provider
28 self._initialized = False
30 async def _ensure_table_exists(self) -> None:
31 """Ensure admin_users table exists (create if needed)."""
32 if self._initialized:
33 return
35 try:
36 # Table check (supports both Postgres and SQLite)
37 db_type = getattr(self.db_provider, "database_type", "") or ""
38 exists = False
40 if db_type.lower() in ("postgres", "postgresql"):
41 check_sql = "SELECT EXISTS (SELECT FROM information_schema.tables WHERE table_schema = 'public' AND table_name = 'admin_users')"
42 result = await self.db_provider.execute_query(check_sql, [])
43 if hasattr(result, "rows") and result.rows:
44 exists = result.rows[0].get("exists", False)
45 elif isinstance(result, list) and result:
46 exists = result[0].get("exists", False)
47 else:
48 # SQLite fallback
49 check_sql = "SELECT name FROM sqlite_master WHERE type='table' AND name='admin_users'"
50 result = await self.db_provider.execute_query(check_sql, [])
51 # Direct check if result has any rows
52 if hasattr(result, "rows"):
53 exists = len(result.rows) > 0
54 elif isinstance(result, list):
55 exists = len(result) > 0
56 else:
57 exists = bool(result)
59 logger.debug("admin_users exists=%s (db_type=%s)", exists, db_type)
61 if not exists:
62 logger.info("Table 'admin_users' not found; creating it...")
63 # Note: UUID in SQLite is TEXT, JSONB is TEXT
64 sql = """
65 CREATE TABLE admin_users (
66 id VARCHAR(255) PRIMARY KEY,
67 name VARCHAR(255) NOT NULL,
68 email VARCHAR(255) UNIQUE NOT NULL,
69 hashed_password TEXT,
70 roles TEXT,
71 permissions TEXT,
72 is_active BOOLEAN DEFAULT true,
73 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
74 updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
75 )
76 """
77 if db_type.lower() in ("postgres", "postgresql"):
78 sql = """
79 CREATE TABLE admin_users (
80 id VARCHAR(255) PRIMARY KEY,
81 name VARCHAR(255) NOT NULL,
82 email VARCHAR(255) UNIQUE NOT NULL,
83 hashed_password TEXT,
84 roles JSONB,
85 permissions JSONB,
86 is_active BOOLEAN DEFAULT true,
87 created_at TIMESTAMPTZ DEFAULT NOW(),
88 updated_at TIMESTAMPTZ DEFAULT NOW()
89 )
90 """
92 await self.db_provider.execute(sql, [])
93 logger.info("✅ admin_users table created successfully")
94 else:
95 logger.debug("Table 'admin_users' already exists")
97 self._initialized = True
98 except Exception as _schema_err: # noqa: BLE001 — schema setup may fail with DB-specific errors; log and propagate
99 logger.exception("Failed to ensure admin_users table exists")
100 raise
102 async def get_admin_count(self) -> int:
103 """Count total admin users."""
104 await self._ensure_table_exists()
105 sql = "SELECT COUNT(*) as count FROM admin_users"
106 result = await self.db_provider.execute_query(sql, [])
108 if hasattr(result, "rows") and result.rows:
109 return result.rows[0].get("count", 0)
110 if isinstance(result, list) and result:
111 return result[0].get("count", 0)
112 if isinstance(result, dict):
113 return result.get("count", 0)
115 return 0
117 async def create_user(
118 self,
119 name: str,
120 email: str,
121 hashed_password: str,
122 roles: list[str] | None = None,
123 permissions: list[str] | None = None,
124 **kwargs, # Accept and ignore extra parameters like 'profile'
125 ) -> Any:
126 await self._ensure_table_exists()
127 admin_id = str(uuid.uuid4())
128 payload = {
129 "id": admin_id,
130 "name": name,
131 "email": email,
132 "hashed_password": hashed_password,
133 "roles": roles or [],
134 "permissions": permissions or [],
135 "is_active": True,
136 }
138 # Attempt an atomic upsert when running against Postgres to avoid races
139 db_type = getattr(self.db_provider, "database_type", "") or ""
140 if db_type.lower() in ("postgres", "postgresql"):
141 # Use RETURNING to fetch created/updated row atomically.
142 # Explicit ::jsonb casts are required because asyncpg cannot
143 # infer the column type for unparameterised JSONB columns
144 # (DataError: expected str, got list).
145 sql = (
146 "INSERT INTO admin_users (id, name, email, hashed_password, roles, permissions, is_active) "
147 "VALUES (?, ?, ?, ?, ?, ?, ?) "
148 "ON CONFLICT (email) DO UPDATE SET "
149 "name = EXCLUDED.name, hashed_password = EXCLUDED.hashed_password, "
150 "roles = EXCLUDED.roles, permissions = EXCLUDED.permissions, is_active = EXCLUDED.is_active, updated_at = NOW() "
151 "RETURNING id, name, email"
152 )
153 params = [
154 admin_id,
155 name,
156 email,
157 hashed_password,
158 roles or [],
159 permissions or [],
160 True,
161 ]
162 try:
163 result = await self.db_provider.execute(sql, params)
164 # db_provider.execute() returns a QueryResult object
165 if hasattr(result, "success") and not result.success:
166 raise RuntimeError(
167 f"UPSERT failed: {getattr(result, 'error_message', 'unknown error')}"
168 )
169 # Extract row from QueryResult.rows
170 row = None
171 if hasattr(result, "rows") and result.rows:
172 row = result.rows[0]
173 elif isinstance(result, list) and result:
174 row = result[0]
175 elif isinstance(result, dict):
176 row = result
178 if row:
180 class _CreatedUser:
181 def __init__(self, uid: str, name: str, email: str) -> None:
182 self.user_id = uid
183 self.name = name
184 self.username = name
185 self.email = email
187 logger.info("Created or updated admin user %s via upsert", name)
188 return _CreatedUser(
189 str(row.get("id")),
190 str(row.get("name") or ""),
191 str(row.get("email") or ""),
192 )
193 except Exception as _upsert_err: # noqa: BLE001 — pragma: no cover - fallback path exercised by tests; DB upsert may raise DB-specific errors
194 # Fall through to non-upsert approach below
195 logger.exception(
196 "Postgres upsert for admin_users failed, falling back to insert"
197 )
199 # Fallback (DB-agnostic): try a manual insert with JSONB casts for Postgres,
200 # falling back to execute_insert for other databases.
201 try:
202 if db_type.lower() in ("postgres", "postgresql"):
203 # Use explicit ::jsonb casts for Postgres
204 fallback_sql = (
205 "INSERT INTO admin_users (id, name, email, hashed_password, roles, permissions, is_active) "
206 "VALUES (?, ?, ?, ?, ?, ?, ?)"
207 )
208 fallback_params = [
209 admin_id,
210 name,
211 email,
212 hashed_password,
213 roles or [],
214 permissions or [],
215 True,
216 ]
217 fallback_result = await self.db_provider.execute(
218 fallback_sql, fallback_params
219 )
220 if hasattr(fallback_result, "success") and not fallback_result.success:
221 raise RuntimeError(
222 f"Fallback INSERT failed: {getattr(fallback_result, 'error_message', 'unknown error')}"
223 )
224 else:
225 await self.db_provider.execute_insert("admin_users", payload)
227 # Return lightweight created object similar shape used by adapter
228 class _CreatedUser: # type: ignore[no-redef]
229 def __init__(self, uid: str, name: str, email: str) -> None:
230 self.user_id = uid
231 self.name = name
232 self.username = name
233 self.email = email
235 logger.info("Created admin user %s via AdminUserStore", name)
236 return _CreatedUser(admin_id, name, email)
237 except Exception as e: # noqa: BLE001 — duplicate-key detection requires inspecting the exception type from any DB driver
238 # Handle duplicate-key errors by resolving existing user and updating if needed
239 err_str = str(e).lower()
240 from lexigram.contracts.exceptions import DuplicateKeyError
242 if (
243 isinstance(e, DuplicateKeyError)
244 or "duplicate key" in err_str
245 or "unique constraint" in err_str
246 ):
247 logger.info(
248 "AdminUserStore.create_user detected existing user for %s; resolving existing user",
249 name,
250 )
251 existing = None
252 try:
253 existing = await self.get_user_by_email(email)
254 except (RuntimeError, ValueError, OSError):
255 existing = None
257 if existing:
258 # Optionally update roles/permissions/hashed_password
259 try:
260 # Populate missing attrs if provided
261 existing_roles = getattr(existing, "roles", []) or []
262 if existing_roles != (roles or []):
263 existing.roles = roles or []
264 existing_perms = getattr(existing, "permissions", []) or []
265 if existing_perms != (permissions or []):
266 existing.permissions = permissions or []
268 existing_hash = getattr(existing, "hashed_password", None)
269 if hashed_password and existing_hash != hashed_password:
270 existing.hashed_password = hashed_password
271 await self.update_user(existing)
272 logger.info(
273 "Updated existing admin user %s after duplicate create",
274 name,
275 )
276 except BaseException:
277 logger.exception(
278 "Failed to update existing admin user %s after duplicate create",
279 name,
280 )
282 # Return lightweight object
283 class _ExistingUser:
284 def __init__(self, uid: str, name: str, email: str) -> None:
285 self.user_id = uid
286 self.name = name
287 self.email = email
289 return _ExistingUser(
290 str(
291 getattr(
292 existing, "user_id", getattr(existing, "id", None) or ""
293 )
294 ),
295 str(existing.name or ""),
296 str(existing.email or ""),
297 )
299 # Re-raise if it's some other error
300 logger.exception(
301 "DirectSQLAdminUserStore.create_user failed for %s",
302 name,
303 )
304 raise
306 # Also need helper for email lookup if we use it in logic
307 async def get_user_by_email(self, email: str) -> Any | None:
308 await self._ensure_table_exists()
309 from lexigram.logging import get_logger
311 logger = get_logger(__name__)
313 sql = "SELECT * FROM admin_users WHERE email = ?"
314 result = await self.db_provider.execute_query(sql, [email])
315 logger.debug(
316 "get_user_by_email result type: %s, result: %s",
317 type(result),
318 result,
319 )
321 row = None
322 if hasattr(result, "rows") and result.rows:
323 row = result.rows[0]
324 elif hasattr(result, "fetchone"):
325 row = result.fetchone()
326 elif isinstance(result, dict):
327 row = result
328 elif isinstance(result, list) and result:
329 row = result[0] if result else None
331 logger.debug("Parsed row: %s", row)
333 if not row:
334 logger.debug("No user found with email: %s", email)
335 return None
337 logger.debug(
338 "Found user by email: %s, is_active: %s, has_password: %s",
339 email,
340 row.get("is_active"),
341 bool(row.get("hashed_password")),
342 )
344 class _UserObj:
345 def __init__(self, row: dict[str, Any]) -> None:
346 self.user_id = str(row.get("id") or row.get("user_id"))
347 self.name = row.get("name")
348 self.email = row.get("email")
349 # Add attributes needed for update
350 self.roles = row.get("roles", [])
351 self.permissions = row.get("permissions", [])
352 self.hashed_password = row.get("hashed_password")
353 self.is_active = row.get("is_active")
355 def record_login(self) -> Any:
356 """Record login - no-op for admin users"""
358 return _UserObj(row)
360 async def get_user_by_id(self, user_id: str) -> Any | None:
361 await self._ensure_table_exists()
362 from lexigram.logging import get_logger
364 logger = get_logger(__name__)
366 sql = "SELECT * FROM admin_users WHERE id = ?"
367 result = await self.db_provider.execute_query(sql, [user_id])
368 row = None
369 if hasattr(result, "rows") and result.rows:
370 row = result.rows[0]
371 elif hasattr(result, "fetchone"):
372 row = result.fetchone()
373 elif isinstance(result, dict):
374 row = result
375 elif isinstance(result, list) and result:
376 row = result[0]
378 if not row:
379 logger.debug("No user found with id: %s", user_id)
380 return None
382 logger.debug(
383 "Found user by id: %s, is_active: %s",
384 user_id,
385 row.get("is_active"),
386 )
388 class _UserObj:
389 def __init__(self, row: dict[str, Any]) -> None:
390 self.user_id = str(row.get("id") or row.get("user_id"))
391 self.name = row.get("name")
392 self.email = row.get("email")
393 self.roles = row.get("roles", [])
394 self.permissions = row.get("permissions", [])
395 self.hashed_password = row.get("hashed_password")
396 self.is_active = row.get("is_active")
398 def record_login(self) -> Any:
399 """Record login - no-op for admin users"""
401 return _UserObj(row)
403 async def update_user(self, user: Any) -> None:
404 await self._ensure_table_exists()
405 payload = {
406 "name": user.name,
407 "email": user.email,
408 "hashed_password": getattr(user, "hashed_password", None),
409 "roles": getattr(user, "roles", []),
410 "permissions": getattr(user, "permissions", []),
411 "is_active": getattr(user, "is_active", True),
412 }
413 await self.db_provider.execute_update(
414 "admin_users",
415 payload,
416 "id = ?",
417 [user.user_id],
418 )
420 async def authenticate(self, email: str, password: str) -> Any | None:
421 """Authenticate an admin user by email and bcrypt-hashed password.
423 Args:
424 email: Email address to look up.
425 password: Plain-text password to verify against the stored hash.
427 Returns:
428 User object when credentials are valid and account is active,
429 ``None`` otherwise.
430 """
431 user = await self.get_user_by_email(email)
432 if not user:
433 return None
434 if not getattr(user, "is_active", True):
435 return None
436 hashed = getattr(user, "hashed_password", None)
437 if not hashed:
438 return None
439 try:
440 import bcrypt
442 hashed_bytes = hashed.encode("utf-8") if isinstance(hashed, str) else hashed
443 if bcrypt.checkpw(password.encode("utf-8"), hashed_bytes):
444 return user
445 except (ValueError, TypeError) as exc:
446 logger.warning(
447 "authenticate.password_check_failed", email=email, error=str(exc)
448 )
449 return None
451 async def delete_user(self, user_id: str) -> None:
452 await self._ensure_table_exists()
453 await self.db_provider.execute_delete("admin_users", "id = ?", [user_id])
455 async def get_by_id(self, admin_id: str) -> Any | None:
456 """Alias for get_user_by_id — used by AdminAuthMiddleware."""
457 return await self.get_user_by_id(admin_id)