Coverage for src / lexigram / admin / auth / store / session_sql.py: 26%
89 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-13 22:14 +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
15if TYPE_CHECKING:
16 from datetime import datetime
18from lexigram.di.decorators import inject
19from lexigram.logging import get_logger
21logger = get_logger(__name__)
24@inject
25class AdminSessionSqlRepository:
26 """SQL-backed repository for admin session persistence.
28 Owns all DDL (table/index creation) and DML (CRUD) for the
29 ``admin_sessions`` table. The constructor accepts any object that
30 satisfies ``DatabaseProviderProtocol`` from ``lexigram-contracts``; the
31 type annotation is kept as ``Any`` here to avoid a circular import — the
32 contract is enforced at DI-wiring time.
34 Implements the ``SessionRepositoryProtocol`` protocol (structural subtyping).
35 """
37 _TABLE = "admin_sessions"
39 def __init__(self, db_provider: DatabaseProviderProtocol) -> None:
40 """Initialise with a resolved database provider.
42 Args:
43 db_provider: Framework database provider that exposes
44 ``execute_insert``, ``execute_query``, and ``execute``.
45 """
46 self._db = db_provider
47 self._initialized = False
49 # ------------------------------------------------------------------
50 # Schema bootstrap (DDL)
51 # ------------------------------------------------------------------
53 async def ensure_schema(self) -> None:
54 """Create the ``admin_sessions`` table and indexes if absent.
56 Safe to call multiple times — the check is idempotent after the first
57 successful run. Raises on any unexpected DDL failure so callers
58 surface the problem rather than silently skipping session persistence.
59 """
60 if self._initialized:
61 return
63 try:
64 db_type = (getattr(self._db, "database_type", "") or "").lower()
65 exists = await self._table_exists(db_type)
67 if not exists:
68 logger.info("Creating %s table…", self._TABLE)
69 await self._db.execute(self._create_table_sql(db_type), [])
70 logger.info("✅ %s table created", self._TABLE)
71 await self._create_indexes()
73 self._initialized = True
75 except Exception as _schema_err: # noqa: BLE001 — schema setup may fail with DB-specific errors; log and propagate
76 logger.exception("Failed to initialise %s schema", self._TABLE)
77 raise
79 async def _table_exists(self, db_type: str) -> bool:
80 if db_type in ("postgres", "postgresql"):
81 sql = (
82 "SELECT EXISTS ("
83 " SELECT FROM information_schema.tables"
84 " WHERE table_schema = 'public'"
85 f" AND table_name = '{self._TABLE}'"
86 ")"
87 )
88 result = await self._db.execute_query(sql, [])
89 if hasattr(result, "rows") and result.rows:
90 return bool(result.rows[0].get("exists", False))
91 if isinstance(result, list) and result:
92 return bool(result[0].get("exists", False))
93 return False
95 # SQLite fallback
96 sql = (
97 "SELECT name FROM sqlite_master "
98 f"WHERE type='table' AND name='{self._TABLE}'"
99 )
100 result = await self._db.execute_query(sql, [])
101 if hasattr(result, "rows"):
102 return len(result.rows) > 0
103 if isinstance(result, list):
104 return len(result) > 0
105 return bool(result)
107 @staticmethod
108 def _create_table_sql(db_type: str) -> str:
109 if db_type in ("postgres", "postgresql"):
110 return """
111 CREATE TABLE admin_sessions (
112 session_id VARCHAR(255) PRIMARY KEY,
113 admin_id VARCHAR(255) NOT NULL,
114 device_id VARCHAR(255),
115 ip_address VARCHAR(45),
116 user_agent TEXT,
117 fingerprint JSONB,
118 fingerprint_sig VARCHAR(64),
119 is_active BOOLEAN NOT NULL DEFAULT true,
120 expires_at TIMESTAMPTZ,
121 created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
122 last_active_at TIMESTAMPTZ DEFAULT NOW(),
123 FOREIGN KEY (admin_id)
124 REFERENCES admin_users(id) ON DELETE CASCADE
125 )
126 """
127 return """
128 CREATE TABLE admin_sessions (
129 session_id VARCHAR(255) PRIMARY KEY,
130 admin_id VARCHAR(255) NOT NULL,
131 device_id VARCHAR(255),
132 ip_address VARCHAR(45),
133 user_agent TEXT,
134 fingerprint TEXT,
135 fingerprint_sig TEXT,
136 is_active BOOLEAN NOT NULL DEFAULT 1,
137 expires_at TIMESTAMP,
138 created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
139 last_active_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
140 )
141 """
143 async def _create_indexes(self) -> None:
144 indexes = [
145 f"CREATE INDEX ix_{self._TABLE}_admin_id ON {self._TABLE}(admin_id)",
146 f"CREATE INDEX ix_{self._TABLE}_device_id ON {self._TABLE}(device_id)",
147 f"CREATE INDEX ix_{self._TABLE}_is_active ON {self._TABLE}(is_active)",
148 f"CREATE INDEX ix_{self._TABLE}_expires_at ON {self._TABLE}(expires_at)",
149 ]
150 for sql in indexes:
151 try:
152 await self._db.execute(sql, [])
153 except (RuntimeError, ValueError, OSError) as exc:
154 logger.debug("Index creation skipped: %s", exc)
155 logger.info("✅ %s indexes created", self._TABLE)
157 # ------------------------------------------------------------------
158 # SessionRepositoryProtocol protocol implementation
159 # ------------------------------------------------------------------
161 async def insert(self, payload: dict[str, Any]) -> None:
162 """Persist a new session record.
164 Args:
165 payload: Field/value mapping (session_id, admin_id, device_id,
166 ip_address, user_agent, fingerprint, expires_at, …).
167 """
168 await self.ensure_schema()
169 await self._db.execute_insert(self._TABLE, payload)
171 async def find_active(self, session_id: str) -> dict[str, Any] | None:
172 """Return the row for an active session, or ``None`` if absent/inactive.
174 Args:
175 session_id: Opaque session identifier.
177 Returns:
178 Raw row dict, or ``None``.
179 """
180 await self.ensure_schema()
181 sql = f"SELECT * FROM {self._TABLE} WHERE session_id = ? AND is_active = TRUE"
182 result = await self._db.execute_query(sql, [session_id])
184 rows = self._extract_rows(result)
185 return dict(rows[0]) if rows else None
187 async def find_active_by_user(
188 self,
189 user_id: str,
190 cutoff: datetime,
191 ) -> list[dict[str, Any]]:
192 """Return all non-expired, active sessions for a user.
194 Args:
195 user_id: Owner identifier.
196 cutoff: Sessions expiring at-or-before this timestamp are excluded.
198 Returns:
199 List of raw row dicts ordered by ``last_active_at`` descending.
200 """
201 await self.ensure_schema()
202 sql = (
203 f"SELECT * FROM {self._TABLE} "
204 "WHERE admin_id = ? AND is_active = TRUE AND expires_at > ? "
205 "ORDER BY last_active_at DESC"
206 )
207 result = await self._db.execute_query(sql, [user_id, cutoff])
208 return [dict(r) for r in self._extract_rows(result)]
210 async def revoke(self, session_id: str) -> None:
211 """Deactivate a single session.
213 Args:
214 session_id: Session to revoke.
215 """
216 await self.ensure_schema()
217 sql = f"UPDATE {self._TABLE} SET is_active = FALSE WHERE session_id = ?"
218 await self._db.execute(sql, (session_id,))
220 async def revoke_all(self, user_id: str) -> None:
221 """Deactivate every active session owned by a user.
223 Args:
224 user_id: Owner whose sessions are to be revoked.
225 """
226 await self.ensure_schema()
227 sql = f"UPDATE {self._TABLE} SET is_active = FALSE WHERE admin_id = ?"
228 await self._db.execute(sql, (user_id,))
230 async def update_activity(self, session_id: str, now: datetime) -> None:
231 """Refresh the ``last_active_at`` timestamp for an active session.
233 Args:
234 session_id: Session to touch.
235 now: Current UTC timestamp to persist.
236 """
237 await self.ensure_schema()
238 sql = (
239 f"UPDATE {self._TABLE} "
240 "SET last_active_at = ? "
241 "WHERE session_id = ? AND is_active = TRUE"
242 )
243 await self._db.execute(sql, (now, session_id))
245 # ------------------------------------------------------------------
246 # Internal helpers
247 # ------------------------------------------------------------------
249 @staticmethod
250 def _extract_rows(result: Any) -> list[Any]:
251 """Normalise heterogeneous query result shapes into a plain list."""
252 if hasattr(result, "rows"):
253 return list(result.rows)
254 if isinstance(result, list):
255 return result
256 return []