53e0065e3e
配置 SSOT（TOML + .env） 统一错误契约 Auth 与事务边界 Redis / Celery 可靠性:业务 Redis（DB/0）与 Celery broker/backend（DB/1）显式拆分；连接池、sync client 可观测性（OpenTelemetry + LGTM）
43 lines
1.3 KiB
Python
43 lines
1.3 KiB
Python
"""Map users-table unique constraint violations to auth error codes."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
_PHONE_CONSTRAINT_MARKERS = frozenset({"phone", "ix_users_phone", "users_phone_key"})
|
|
_EMAIL_CONSTRAINT_MARKERS = frozenset({"email", "users_email_key", "ix_users_email"})
|
|
|
|
|
|
def _constraint_text(exc: IntegrityError) -> str:
|
|
orig = exc.orig
|
|
if orig is None:
|
|
return str(exc).lower()
|
|
diag = getattr(orig, "diag", None)
|
|
if diag is not None:
|
|
name = getattr(diag, "constraint_name", None)
|
|
if name:
|
|
return str(name).lower()
|
|
return str(orig).lower()
|
|
|
|
|
|
def _matches(markers: frozenset[str], text: str) -> bool:
|
|
return any(marker in text for marker in markers)
|
|
|
|
|
|
def user_integrity_auth_code(
|
|
exc: IntegrityError,
|
|
*,
|
|
phone_conflict: str,
|
|
) -> str | None:
|
|
"""Return auth internal code for a users-table unique violation, or None."""
|
|
text = _constraint_text(exc)
|
|
if _matches(_PHONE_CONSTRAINT_MARKERS, text):
|
|
return phone_conflict
|
|
if _matches(_EMAIL_CONSTRAINT_MARKERS, text):
|
|
return "EMAIL_EXISTS"
|
|
return None
|
|
|
|
|
|
def is_user_phone_unique_violation(exc: IntegrityError) -> bool:
|
|
return _matches(_PHONE_CONSTRAINT_MARKERS, _constraint_text(exc))
|