Coverage for src/lexigram/admin/auth/store/session_sql.py: 65%
107 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 14:56 +0800
1"""SQL-backed implementation of SessionRepositoryProtocol for admin sessions.
3This is the *only* module in ``lexigram-admin`` that issues raw SQL against
4the ``admin_sessions`` table. The session-management service layer depends
5on the ``SessionRepositoryProtocol`` protocol from ``lexigram-contracts`` and never
6on this class directly.
7"""
9from __future__ import annotations
11from typing import TYPE_CHECKING, Any
13from lexigram.contracts.data import DatabaseProviderProtocol, Table
14from lexigram.serialization import dumps_str, loads_str
16if TYPE_CHECKING:
17 from datetime import datetime
19from lexigram.di.decorators import inject
20from lexigram.logging import get_logger
22logger = get_logger(__name__)
25@inject
26class AdminSessionSqlRepository:
27 """SQL-backed repository for admin session persistence.
29 Owns all DDL (table/index creation) and DML (CRUD) for the
30 ``admin_sessions`` table. The constructor accepts any object that
31 satisfies ``DatabaseProviderProtocol`` from ``lexigram-contracts``; the
32 type annotation is kept as ``Any`` here to avoid a circular import — the
33 contract is enforced at DI-wiring time.
35 Implements the ``SessionRepositoryProtocol`` protocol (structural subtyping).
36 """
38 _TABLE = Table("admin_sessions")
40 def __init__(self, db_provider: DatabaseProviderProtocol) -> None:
41 """Initialise with a resolved database provider.
43 Args:
44 db_provider: Framework database provider that exposes
45 ``execute_insert``, ``execute_query``, and ``execute``.
46 """
47 self._db = db_provider
48 self._initialized = False
50 # ------------------------------------------------------------------
51 # Schema bootstrap (DDL)
52 # ------------------------------------------------------------------
54 async def ensure_schema(self) -> None:
55 """Create the ``admin_sessions`` table and indexes if absent.
57 Safe to call multiple times — the check is idempotent after the first
58 successful run. Raises on any unexpected DDL failure so callers
59 surface the problem rather than silently skipping session persistence.
60 """
61 if self._initialized:
62 return
64 try:
65 db_type = (getattr(self._db, "database_type", "") or "").lower()
66 exists = await self._table_exists(db_type)
68 if not exists:
69 logger.info("Creating %s table…", self._TABLE)
70 await self._db.execute(self._create_table_sql(db_type), [])
71 logger.info("✅ %s table created", self._TABLE)
72 await self._create_indexes()
74 self._initialized = True
76 except Exception as _schema_err: # noqa: BLE001 — schema setup may fail with DB-specific errors; log and propagate
77 logger.exception("Failed to initialise %s schema", self._TABLE)
78 raise
80 async def _table_exists(self, db_type: str) -> bool:
81 if db_type in ("postgres", "postgresql"):
82 sql = (
83 "SELECT EXISTS (" # noqa: S608 — table name is constant class attr Table("admin_sessions"), never user input
84 " SELECT FROM information_schema.tables"
85 " WHERE table_schema = 'public'"
86 f" AND table_name = '{self._TABLE.name}'"
87 ")"
88 )
89 result = await self._db.execute_query(sql, [])
90 if hasattr(result, "rows") and result.rows:
91 return bool(result.rows[0].get("exists", False))
92 if isinstance(result, list) and result:
93 return bool(result[0].get("exists", False))
94 return False
96 # SQLite fallback
97 sql = (
98 "SELECT name FROM sqlite_master " # noqa: S608 — table name is constant class attr Table("admin_sessions"), never user input
99 f"WHERE type='table' AND name='{self._TABLE.name}'"
100 )
101 result = await self._db.execute_query(sql, [])
102 if hasattr(result, "rows"):
103 return len(result.rows) > 0
104 if isinstance(result, list):
105 return len(result) > 0
106 return bool(result)
108 @staticmethod
109 def _create_table_sql(db_type: str) -> str:
110 if db_type in ("postgres", "postgresql"):
111 return """
112 CREATE TABLE admin_sessions (
113 session_id VARCHAR(255) PRIMARY KEY,
114 admin_id VARCHAR(255) NOT NULL,
115 device_id VARCHAR(255),
116 ip_address VARCHAR(45),
117 user_agent TEXT,
118 fingerprint JSONB,
119 fingerprint_sig VARCHAR(64),
120 is_active BOOLEAN NOT NULL DEFAULT true,
121 expires_at TIMESTAMPTZ,
122 created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
123 last_active_at TIMESTAMPTZ DEFAULT NOW(),
124 FOREIGN KEY (admin_id)
125 REFERENCES admin_users(id) ON DELETE CASCADE
126 )
127 """
128 return """
129 CREATE TABLE admin_sessions (
130 session_id VARCHAR(255) PRIMARY KEY,
131 admin_id VARCHAR(255) NOT NULL,
132 device_id VARCHAR(255),
133 ip_address VARCHAR(45),
134 user_agent TEXT,
135 fingerprint TEXT,
136 fingerprint_sig TEXT,
137 is_active BOOLEAN NOT NULL DEFAULT 1,
138 expires_at TIMESTAMP,
139 created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
140 last_active_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
141 )
142 """
144 async def _create_indexes(self) -> None:
145 indexes = [
146 f"CREATE INDEX ix_{self._TABLE.name}_admin_id ON {self._TABLE}(admin_id)",
147 f"CREATE INDEX ix_{self._TABLE.name}_device_id ON {self._TABLE}(device_id)",
148 f"CREATE INDEX ix_{self._TABLE.name}_is_active ON {self._TABLE}(is_active)",
149 f"CREATE INDEX ix_{self._TABLE.name}_expires_at ON {self._TABLE}(expires_at)",
150 ]
151 for sql in indexes:
152 try:
153 await self._db.execute(sql, [])
154 except (RuntimeError, ValueError, OSError) as exc:
155 logger.debug("Index creation skipped: %s", exc)
156 logger.info("✅ %s indexes created", self._TABLE)
158 # ------------------------------------------------------------------
159 # SessionRepositoryProtocol protocol implementation
160 # ------------------------------------------------------------------
162 async def insert(self, payload: dict[str, Any]) -> None:
163 """Persist a new session record.
165 Args:
166 payload: Field/value mapping (session_id, admin_id, device_id,
167 ip_address, user_agent, fingerprint, expires_at, …).
168 """
169 await self.ensure_schema()
170 payload = dict(payload)
171 fingerprint = payload.get("fingerprint")
172 if isinstance(fingerprint, dict):
173 db_type = (getattr(self._db, "database_type", "") or "").lower()
174 if db_type in ("sqlite", "sqlite3", "memory"):
175 payload["fingerprint"] = dumps_str(fingerprint)
176 await self._db.execute_insert(self._TABLE.name, payload)
178 async def find_active(self, session_id: str) -> dict[str, Any] | None:
179 """Return the row for an active session, or ``None`` if absent/inactive.
181 Args:
182 session_id: Opaque session identifier.
184 Returns:
185 Raw row dict, or ``None``.
186 """
187 await self.ensure_schema()
188 sql = f"SELECT * FROM {self._TABLE} WHERE session_id = ? AND is_active = TRUE" # noqa: S608 — table name is constant class attr Table("admin_sessions"), never user input
189 result = await self._db.execute_query(sql, [session_id])
191 rows = self._extract_rows(result)
192 if not rows:
193 return None
194 return self._decode_fingerprint(dict(rows[0]))
196 async def find_active_by_user(
197 self,
198 user_id: str,
199 cutoff: datetime,
200 ) -> list[dict[str, Any]]:
201 """Return all non-expired, active sessions for a user.
203 Args:
204 user_id: Owner identifier.
205 cutoff: Sessions expiring at-or-before this timestamp are excluded.
207 Returns:
208 List of raw row dicts ordered by ``last_active_at`` descending.
209 """
210 await self.ensure_schema()
211 sql = (
212 f"SELECT * FROM {self._TABLE} " # noqa: S608 — table name is constant class attr Table("admin_sessions"), never user input
213 "WHERE admin_id = ? AND is_active = TRUE AND expires_at > ? "
214 "ORDER BY last_active_at DESC"
215 )
216 result = await self._db.execute_query(sql, [user_id, cutoff])
217 return [self._decode_fingerprint(dict(r)) for r in self._extract_rows(result)]
219 async def revoke(self, session_id: str) -> None:
220 """Deactivate a single session.
222 Args:
223 session_id: Session to revoke.
224 """
225 await self.ensure_schema()
226 sql = f"UPDATE {self._TABLE} SET is_active = FALSE WHERE session_id = ?" # noqa: S608 — table name is constant class attr Table("admin_sessions"), never user input
227 await self._db.execute(sql, (session_id,))
229 async def revoke_all(self, user_id: str) -> None:
230 """Deactivate every active session owned by a user.
232 Args:
233 user_id: Owner whose sessions are to be revoked.
234 """
235 await self.ensure_schema()
236 sql = f"UPDATE {self._TABLE} SET is_active = FALSE WHERE admin_id = ?" # noqa: S608 — table name is constant class attr Table("admin_sessions"), never user input
237 await self._db.execute(sql, (user_id,))
239 async def update_activity(self, session_id: str, now: datetime) -> None:
240 """Refresh the ``last_active_at`` timestamp for an active session.
242 Args:
243 session_id: Session to touch.
244 now: Current UTC timestamp to persist.
245 """
246 await self.ensure_schema()
247 sql = (
248 f"UPDATE {self._TABLE} " # noqa: S608 — table name is constant class attr Table("admin_sessions"), never user input
249 "SET last_active_at = ? "
250 "WHERE session_id = ? AND is_active = TRUE"
251 )
252 await self._db.execute(sql, (now, session_id))
254 # ------------------------------------------------------------------
255 # Internal helpers
256 # ------------------------------------------------------------------
258 @staticmethod
259 def _extract_rows(result: Any) -> list[Any]:
260 """Normalise heterogeneous query result shapes into a plain list."""
261 if hasattr(result, "rows"):
262 return list(result.rows)
263 if isinstance(result, list):
264 return result
265 return []
267 @staticmethod
268 def _decode_fingerprint(row: dict[str, Any]) -> dict[str, Any]:
269 """Deserialize a string-stored fingerprint back into a dict for SQLite."""
270 fingerprint = row.get("fingerprint")
271 if isinstance(fingerprint, str):
272 try:
273 row["fingerprint"] = loads_str(fingerprint)
274 except ValueError:
275 pass
276 return row