76 lines
2.4 KiB
Python
76 lines
2.4 KiB
Python
# Oracle 异常映射：ORA 码、约束名、固定对外文案。
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
|
||
_ORA_CODE_RE = re.compile(r"ORA-(\d+)", re.I)
|
||
_CONSTRAINT_RE = re.compile(
|
||
r"unique constraint \((?:[\w$#]+\.)?([\w$#]+)\)",
|
||
re.I,
|
||
)
|
||
|
||
SALES_OMS_ID_CONSTRAINT = "UK_RPA_SGMD_OMS_ID"
|
||
CONTRACT_FID_CONSTRAINTS = frozenset({"PK_RPA_PCD"})
|
||
|
||
|
||
# 数据库连不上、未配置或 thick client 初始化失败。
|
||
#
|
||
# 对外消息固定为「数据库不可用」，不带 DSN、库路径或 ORA 原文。
|
||
class RepoUnavailable(Exception):
|
||
# 记下对外文案。默认「数据库不可用」，不要把 ORA 原文传进来。
|
||
def __init__(self, message: str = "数据库不可用"):
|
||
self.message = message
|
||
super().__init__(message)
|
||
|
||
|
||
# 取出异常上的 Oracle 错误号，没有则返回 None。
|
||
def ora_code(exc: BaseException) -> int | None:
|
||
if exc.args:
|
||
info = exc.args[0]
|
||
code = getattr(info, "code", None)
|
||
if code is not None:
|
||
return int(code)
|
||
code = getattr(exc, "code", None)
|
||
if code is not None:
|
||
return int(code)
|
||
match = _ORA_CODE_RE.search(str(exc))
|
||
if match:
|
||
return int(match.group(1))
|
||
return None
|
||
|
||
|
||
# 从 ORA-00001 报文里取出约束名（不含 schema），没有则返回 None。
|
||
def ora_constraint(exc: BaseException) -> str | None:
|
||
chunks: list[str] = [str(exc)]
|
||
if exc.args:
|
||
info = exc.args[0]
|
||
message = getattr(info, "message", None)
|
||
if message:
|
||
chunks.append(str(message))
|
||
for text in chunks:
|
||
match = _CONSTRAINT_RE.search(text)
|
||
if match:
|
||
return match.group(1)
|
||
return None
|
||
|
||
|
||
# 是否为表 1 oms_id 唯一约束冲突（接口 1 DUPLICATE）。
|
||
#
|
||
# 只有约束名是 uk_rpa_sgmd_oms_id 才算。其它唯一约束即使是 ORA-00001 也返回 False。
|
||
def is_sales_oms_id_duplicate(exc: BaseException) -> bool:
|
||
if ora_code(exc) != 1:
|
||
return False
|
||
name = ora_constraint(exc)
|
||
return bool(name) and name.upper() == SALES_OMS_ID_CONSTRAINT
|
||
|
||
|
||
# 是否为表 3 fid 主键冲突（接口 3 DUPLICATE）。
|
||
#
|
||
# 目前对应 pk_rpa_pcd。未知唯一约束返回 False。
|
||
def is_contract_fid_duplicate(exc: BaseException) -> bool:
|
||
if ora_code(exc) != 1:
|
||
return False
|
||
name = ora_constraint(exc)
|
||
return bool(name) and name.upper() in CONTRACT_FID_CONSTRAINTS
|