Coverage for src/lexigram/admin/auth/store/login_attempt_sql.py: 0%
64 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-24 23:18 +0800
1"""SQL-backed implementation of AdminLoginAttemptStoreProtocol.
3Owns all DDL and DML for the ``admin_login_attempts`` table. The service
4layer depends only on ``AdminLoginAttemptStoreProtocol`` from
5``lexigram.admin.auth.protocols`` — never on this class directly.
6"""
8from __future__ import annotations
10from typing import TYPE_CHECKING, Any
11import uuid
13from lexigram.admin.sql_dialect import is_postgres, since_expr
14from lexigram.contracts.data import DatabaseProviderProtocol
15from lexigram.di.decorators import inject
16from lexigram.logging import get_logger
18if TYPE_CHECKING:
19 from lexigram.admin.auth.types import AdminLoginAttempt
21logger = get_logger(__name__)
23_TABLE = "admin_login_attempts"
26@inject
27class AdminLoginAttemptSqlStore:
28 """SQL-backed store for admin login attempt records.
30 Implements ``AdminLoginAttemptStoreProtocol`` via structural subtyping.
31 Manages the ``admin_login_attempts`` table including DDL bootstrap,
32 failure counting for rate limiting, and attempt persistence.
33 """
35 def __init__(self, db: DatabaseProviderProtocol) -> None:
36 """Initialise with a resolved database provider.
38 Args:
39 db: Framework database provider exposing ``execute`` and
40 ``execute_query``.
41 """
42 self._db = db
43 self._initialized = False
45 # ------------------------------------------------------------------
46 # Schema bootstrap (DDL)
47 # ------------------------------------------------------------------
49 async def ensure_schema(self) -> None:
50 """Create table and indexes if they do not exist.
52 Safe to call multiple times — idempotent after the first successful
53 run. Uses ``CREATE TABLE IF NOT EXISTS`` and
54 ``CREATE INDEX IF NOT EXISTS`` so concurrent callers are safe.
56 Raises:
57 Exception: Propagates any unexpected DDL failure so the caller
58 surfaces the problem rather than silently skipping persistence.
59 """
60 if self._initialized:
61 return
63 try:
64 if is_postgres(self._db):
65 create_sql = """
66 CREATE TABLE IF NOT EXISTS admin_login_attempts (
67 id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
68 email VARCHAR(255) NOT NULL,
69 ip_address VARCHAR(45) NOT NULL,
70 user_agent TEXT NOT NULL DEFAULT '',
71 success BOOLEAN NOT NULL DEFAULT FALSE,
72 failure_reason VARCHAR(50),
73 attempted_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
74 )
75 """
76 else:
77 create_sql = """
78 CREATE TABLE IF NOT EXISTS admin_login_attempts (
79 id TEXT PRIMARY KEY,
80 email VARCHAR(255) NOT NULL,
81 ip_address VARCHAR(45) NOT NULL,
82 user_agent TEXT NOT NULL DEFAULT '',
83 success BOOLEAN NOT NULL DEFAULT FALSE,
84 failure_reason VARCHAR(50),
85 attempted_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
86 )
87 """
88 await self._db.execute(create_sql, [])
89 await self._db.execute(
90 """
91 CREATE INDEX IF NOT EXISTS idx_admin_login_attempts_email_attempted_at
92 ON admin_login_attempts(email, attempted_at DESC)
93 """,
94 [],
95 )
96 await self._db.execute(
97 """
98 CREATE INDEX IF NOT EXISTS idx_admin_login_attempts_ip_attempted_at
99 ON admin_login_attempts(ip_address, attempted_at DESC)
100 """,
101 [],
102 )
103 self._initialized = True
104 logger.info("✅ %s schema ready", _TABLE)
105 except Exception as _schema_err: # noqa: BLE001 — DDL may raise DB-specific errors; log and propagate
106 logger.exception("Failed to initialise %s schema", _TABLE)
107 raise
109 # ------------------------------------------------------------------
110 # DML
111 # ------------------------------------------------------------------
113 async def insert(self, attempt: AdminLoginAttempt) -> None:
114 """Insert a login attempt record.
116 Generates a fresh UUID when ``attempt.id`` is empty so callers may
117 pass an uninitialised id without failing the NOT NULL constraint.
119 Args:
120 attempt: Login attempt data to persist.
121 """
122 await self.ensure_schema()
123 attempt_id = attempt.id or str(uuid.uuid4())
124 sql = (
125 "INSERT INTO admin_login_attempts "
126 "(id, email, ip_address, user_agent, success, failure_reason, attempted_at) "
127 "VALUES (?, ?, ?, ?, ?, ?, ?)"
128 )
129 await self._db.execute(
130 sql,
131 (
132 attempt_id,
133 attempt.email,
134 attempt.ip_address,
135 attempt.user_agent,
136 attempt.success,
137 attempt.failure_reason,
138 attempt.attempted_at,
139 ),
140 )
141 logger.debug(
142 "login_attempt.inserted",
143 email=attempt.email,
144 success=attempt.success,
145 failure_reason=attempt.failure_reason,
146 )
148 async def count_recent_failures(self, email: str, since_seconds: int) -> int:
149 """Count failed login attempts for email within the last N seconds.
151 Args:
152 email: Email address to query.
153 since_seconds: Look-back window in seconds.
155 Returns:
156 Number of failed attempts within the window.
157 """
158 await self.ensure_schema()
159 sql = (
160 "SELECT COUNT(*) AS count FROM admin_login_attempts " # noqa: S608 — since_expr(..., int-cast seconds) builds fixed dialect expression only
161 "WHERE email = ? AND success = FALSE "
162 f"AND attempted_at > {since_expr(self._db, since_seconds)}"
163 )
164 result = await self._db.execute_query(sql, [email])
165 count = self._extract_count(result)
166 logger.debug(
167 "login_attempt.count_recent_failures",
168 email=email,
169 since_seconds=since_seconds,
170 count=count,
171 )
172 return count
174 async def count_recent_failures_by_ip(
175 self, ip_address: str, since_seconds: int
176 ) -> int:
177 """Count failed login attempts from IP within the last N seconds.
179 Args:
180 ip_address: Client IP address to query.
181 since_seconds: Look-back window in seconds.
183 Returns:
184 Number of failed attempts within the window.
185 """
186 await self.ensure_schema()
187 sql = (
188 "SELECT COUNT(*) AS count FROM admin_login_attempts " # noqa: S608 — since_expr(..., int-cast seconds) builds fixed dialect expression only
189 "WHERE ip_address = ? AND success = FALSE "
190 f"AND attempted_at > {since_expr(self._db, since_seconds)}"
191 )
192 result = await self._db.execute_query(sql, [ip_address])
193 count = self._extract_count(result)
194 logger.debug(
195 "login_attempt.count_recent_failures_by_ip",
196 ip_address=ip_address,
197 since_seconds=since_seconds,
198 count=count,
199 )
200 return count
202 async def clear_failures(self, email: str) -> None:
203 """Delete all failure records for email.
205 Called immediately after a successful login so subsequent failure
206 counts start from zero.
208 Args:
209 email: Email address whose failure records to remove.
210 """
211 await self.ensure_schema()
212 sql = "DELETE FROM admin_login_attempts WHERE email = ? AND success = FALSE"
213 await self._db.execute(sql, (email,))
214 logger.debug("login_attempt.failures_cleared", email=email)
216 # ------------------------------------------------------------------
217 # Internal helpers
218 # ------------------------------------------------------------------
220 @staticmethod
221 def _extract_count(result: Any) -> int:
222 """Normalise heterogeneous query result shapes into an integer count.
224 Args:
225 result: Raw result returned by ``execute_query``.
227 Returns:
228 The integer value of the ``count`` column, or ``0`` when absent.
229 """
230 if hasattr(result, "rows") and result.rows:
231 return int(result.rows[0].get("count", 0))
232 if isinstance(result, list) and result:
233 row = result[0]
234 if isinstance(row, dict):
235 return int(row.get("count", 0))
236 return 0
239__all__ = ["AdminLoginAttemptSqlStore"]