Coverage for src/lexigram/admin/auth/store/email_verification_sql.py: 0%
67 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-24 23:18 +0800
1"""SQL-backed implementation of AdminEmailVerificationStoreProtocol.
3Owns all DDL and DML for the ``admin_email_verifications`` table. The
4service layer depends only on ``AdminEmailVerificationStoreProtocol`` from
5``lexigram.admin.auth.protocols`` — never on this class directly.
6"""
8from __future__ import annotations
10from datetime import UTC, datetime
11from typing import Any
13from lexigram.admin.sql_dialect import is_postgres, now_expr
14from lexigram.contracts.data import DatabaseProviderProtocol
15from lexigram.di.decorators import inject
16from lexigram.logging import get_logger
18logger = get_logger(__name__)
20_TABLE = "admin_email_verifications"
23def _parse_dt(value: Any) -> datetime | None:
24 """Parse a provider-returned timestamp into a UTC-aware datetime."""
25 if value is None:
26 return None
27 if isinstance(value, datetime):
28 return value
29 parsed = datetime.fromisoformat(str(value))
30 if parsed.tzinfo is None:
31 return parsed.replace(tzinfo=UTC)
32 return parsed
35@inject
36class AdminEmailVerificationSqlStore:
37 """SQL-backed store for admin email verification state.
39 Implements ``AdminEmailVerificationStoreProtocol`` via structural
40 subtyping. Manages the ``admin_email_verifications`` table including
41 DDL bootstrap and single-use token consumption semantics.
42 """
44 def __init__(self, db: DatabaseProviderProtocol) -> None:
45 """Initialise with a resolved database provider.
47 Args:
48 db: Framework database provider exposing ``execute`` and
49 ``execute_query``.
50 """
51 self._db = db
52 self._initialized = False
54 # ------------------------------------------------------------------
55 # Schema bootstrap (DDL)
56 # ------------------------------------------------------------------
58 async def ensure_schema(self) -> None:
59 """Create the verification table if it does not exist (idempotent)."""
60 if self._initialized:
61 return
62 if is_postgres(self._db):
63 create_sql = f"""
64 CREATE TABLE IF NOT EXISTS {_TABLE} (
65 user_id TEXT PRIMARY KEY,
66 email_verified_at TIMESTAMPTZ,
67 token_hash VARCHAR(64),
68 token_expires_at TIMESTAMPTZ,
69 updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
70 )
71 """
72 else:
73 create_sql = f"""
74 CREATE TABLE IF NOT EXISTS {_TABLE} (
75 user_id TEXT PRIMARY KEY,
76 email_verified_at TIMESTAMP,
77 token_hash VARCHAR(64),
78 token_expires_at TIMESTAMP,
79 updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
80 )
81 """
82 await self._db.execute(create_sql, [])
83 await self._db.execute(
84 f"""
85 CREATE INDEX IF NOT EXISTS idx_{_TABLE}_token
86 ON {_TABLE} (token_hash)
87 """,
88 [],
89 )
90 self._initialized = True
92 # ------------------------------------------------------------------
93 # AdminEmailVerificationStoreProtocol
94 # ------------------------------------------------------------------
96 async def is_verified(self, user_id: str) -> bool:
97 """Return True when the user's email is verified (see protocol docs)."""
98 result = await self._db.execute_query(
99 f"SELECT email_verified_at FROM {_TABLE} WHERE user_id = ?", # noqa: S608 — table name is module constant "admin_email_verifications", never user input
100 [user_id],
101 )
102 row = None
103 if hasattr(result, "rows") and result.rows:
104 row = result.rows[0]
105 elif isinstance(result, list) and result:
106 row = result[0]
107 elif isinstance(result, dict):
108 row = result
109 if not row:
110 return False
111 return _parse_dt(row.get("email_verified_at")) is not None
113 async def find_user_by_token_hash(self, token_hash: str) -> str | None:
114 """Look up the user owning an unconsumed token (see protocol docs)."""
115 result = await self._db.execute_query(
116 f"SELECT user_id FROM {_TABLE} " # noqa: S608 — table name is module constant "admin_email_verifications", never user input
117 "WHERE token_hash = ? AND email_verified_at IS NULL",
118 [token_hash],
119 )
120 row = None
121 if hasattr(result, "rows") and result.rows:
122 row = result.rows[0]
123 elif isinstance(result, list) and result:
124 row = result[0]
125 elif isinstance(result, dict):
126 row = result
127 if not row:
128 return None
129 return str(row.get("user_id", ""))
131 async def save_token(
132 self, user_id: str, token_hash: str, expires_at: datetime
133 ) -> None:
134 """Persist (or refresh) the verification token for a user."""
135 await self._db.execute(
136 f"""
137 INSERT INTO {_TABLE} (user_id, token_hash, token_expires_at)
138 VALUES (?, ?, ?)
139 ON CONFLICT (user_id) DO UPDATE SET
140 token_hash = excluded.token_hash,
141 token_expires_at = excluded.token_expires_at,
142 updated_at = {now_expr(self._db)}
143 """, # noqa: S608 — table name is module constant, now_expr yields fixed NOW()/CURRENT_TIMESTAMP
144 [user_id, token_hash, expires_at],
145 )
147 async def consume_token(self, user_id: str, token_hash: str) -> bool:
148 """Atomically verify + consume a token (see protocol docs)."""
149 result = await self._db.execute(
150 f"""
151 UPDATE {_TABLE} SET
152 email_verified_at = {now_expr(self._db)},
153 token_hash = NULL,
154 token_expires_at = NULL,
155 updated_at = {now_expr(self._db)}
156 WHERE user_id = ?
157 AND token_hash = ?
158 AND email_verified_at IS NULL
159 AND token_expires_at > {now_expr(self._db)}
160 """, # noqa: S608 — table name is module constant, now_expr yields fixed NOW()/CURRENT_TIMESTAMP
161 [user_id, token_hash],
162 )
163 row_count = getattr(result, "row_count", None)
164 if row_count is not None:
165 return int(row_count) > 0
166 return True
168 async def clear_token(self, user_id: str) -> None:
169 """Remove the pending verification token for a user."""
170 await self._db.execute(
171 f"""
172 UPDATE {_TABLE} SET
173 token_hash = NULL,
174 token_expires_at = NULL,
175 updated_at = {now_expr(self._db)}
176 WHERE user_id = ?
177 """, # noqa: S608 — table name is module constant, now_expr yields fixed NOW()/CURRENT_TIMESTAMP
178 [user_id],
179 )
182__all__ = ["AdminEmailVerificationSqlStore"]