a0086554b1
六个 service 各自复制的 _is_duplicate_entry（口径已漂移：rating/product 用 _ERR_DUP_ENTRY 常量、其余四处写字面量 1062）收敛为共享函数； is_duplicate_entry 语义与文案注释各处保留，非 1062 完整性错误仍原样 上抛由全局处理器收敛。 Co-Authored-By: Claude Code <noreply@anthropic.com>
19 lines
806 B
Python
19 lines
806 B
Python
"""数据库完整性错误的统一识别与翻译口径。
|
||
|
||
判断依据：pymysql 原生异常 ``orig.args`` 首位是 MySQL 错误码 —— 只有
|
||
1062（ER_DUP_ENTRY，唯一键冲突）是「已存在」语义，可安全转 400 业务
|
||
提示；其余完整性错误（外键 RESTRICT、数据截断等）原样上抛，由全局
|
||
异常处理器收敛 500（程序缺陷或数据异常，均不应伪装成用户错误）。
|
||
"""
|
||
|
||
from sqlalchemy.exc import IntegrityError
|
||
|
||
# MySQL 唯一键冲突错误码（pymysql 原生异常 args 首位）
|
||
ERR_DUP_ENTRY = 1062
|
||
|
||
|
||
def is_duplicate_entry(exc: IntegrityError) -> bool:
|
||
"""判断 IntegrityError 是否为 MySQL 唯一键冲突（错误码 1062）。"""
|
||
args = getattr(exc.orig, "args", ())
|
||
return bool(args) and args[0] == ERR_DUP_ENTRY
|