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. 

2 

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""" 

8 

9from __future__ import annotations 

10 

11from typing import TYPE_CHECKING, Any 

12 

13from lexigram.contracts.data import DatabaseProviderProtocol 

14 

15if TYPE_CHECKING: 

16 from datetime import datetime 

17 

18from lexigram.di.decorators import inject 

19from lexigram.logging import get_logger 

20 

21logger = get_logger(__name__) 

22 

23 

24@inject 

25class AdminSessionSqlRepository: 

26 """SQL-backed repository for admin session persistence. 

27 

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. 

33 

34 Implements the ``SessionRepositoryProtocol`` protocol (structural subtyping). 

35 """ 

36 

37 _TABLE = "admin_sessions" 

38 

39 def __init__(self, db_provider: DatabaseProviderProtocol) -> None: 

40 """Initialise with a resolved database provider. 

41 

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 

48 

49 # ------------------------------------------------------------------ 

50 # Schema bootstrap (DDL) 

51 # ------------------------------------------------------------------ 

52 

53 async def ensure_schema(self) -> None: 

54 """Create the ``admin_sessions`` table and indexes if absent. 

55 

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 

62 

63 try: 

64 db_type = (getattr(self._db, "database_type", "") or "").lower() 

65 exists = await self._table_exists(db_type) 

66 

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() 

72 

73 self._initialized = True 

74 

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 

78 

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 

94 

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) 

106 

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 """ 

142 

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) 

156 

157 # ------------------------------------------------------------------ 

158 # SessionRepositoryProtocol protocol implementation 

159 # ------------------------------------------------------------------ 

160 

161 async def insert(self, payload: dict[str, Any]) -> None: 

162 """Persist a new session record. 

163 

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) 

170 

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. 

173 

174 Args: 

175 session_id: Opaque session identifier. 

176 

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]) 

183 

184 rows = self._extract_rows(result) 

185 return dict(rows[0]) if rows else None 

186 

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. 

193 

194 Args: 

195 user_id: Owner identifier. 

196 cutoff: Sessions expiring at-or-before this timestamp are excluded. 

197 

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)] 

209 

210 async def revoke(self, session_id: str) -> None: 

211 """Deactivate a single session. 

212 

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,)) 

219 

220 async def revoke_all(self, user_id: str) -> None: 

221 """Deactivate every active session owned by a user. 

222 

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,)) 

229 

230 async def update_activity(self, session_id: str, now: datetime) -> None: 

231 """Refresh the ``last_active_at`` timestamp for an active session. 

232 

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)) 

244 

245 # ------------------------------------------------------------------ 

246 # Internal helpers 

247 # ------------------------------------------------------------------ 

248 

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 []