《FastAPI零基础入门与进阶实战》第20篇:消息管理-封装
系列文章目录
《FASTAPI零基础入门与进阶实战》
https://blog.csdn.net/sen_shan/category_12950843.html
第19篇:消息管理https://blog.csdn.net/sen_shan/article/details/151827453?spm=1011.2415.3001.5331
文章目录
目录
系列文章目录
文章目录
前言
预期效果
初始化信息表
env
load_json
config
json
message初始化
项目初始化
项目初始化
运行
封装
信息封装
调用调整
测试
前言
对消息统一封装消息内容，把code→msg替换，同时根据请求头Accept-Language返回中/英/繁体等多语满足国际化、运维、合规三方面的统一消息管理需求。
上章节主要是Models,Schemas,CRUD,Routerg构建，本章节对初始化与消息封装。
预期效果
返回值message实现中文、英文、繁体国际化
"message": "資料已存在：跳過3筆資料"
{"status_code": 200,"status": "success","message": "資料已存在：跳過3筆資料","record_count": 0,"data": {"created": {"count": 0,"details": []},"duplicates": {"count": 3,"details": [{"login_id": "alice1736","user_name": "Alice","role": null,"email": "alice@example.com","person_id": "P001","is_superuser": false,"is_staff": true,"effective_date": "2023-01-01","expiry_date": "2025-12-31","deletion_mark": false,"deletion_reason": null,"deletion_date": null,"deletion_by": null,"create_date": "2025-09-15T16:56:58","creator_by": "231b2c7d-a2eb-47db-a40d-d400f7d250f6","last_updated_date": null,"last_updated_by": null,"id": "bb6b9a96-239b-4956-a2a3-c39d9f6113fe"},{"login_id": "bob4","user_name": "Bob","role": null,"email": "bob@example.com","person_id": "P002","is_superuser": false,"is_staff": false,"effective_date": "2023-02-01","expiry_date": "2025-12-31","deletion_mark": true,"deletion_reason": null,"deletion_date": "2025-09-15T16:59:29","deletion_by": "231b2c7d-a2eb-47db-a40d-d400f7d250f6","create_date": "2025-09-15T16:59:29","creator_by": "231b2c7d-a2eb-47db-a40d-d400f7d250f6","last_updated_date": null,"last_updated_by": null,"id": "eca07122-f4a2-42e7-b436-e7abb134b1d3"},{"login_id": "charlie3","user_name": "Charlie","role": null,"email": "charlie@example.com","person_id": "P003","is_superuser": true,"is_staff": true,"effective_date": "2023-03-01","expiry_date": "2025-12-31","deletion_mark": false,"deletion_reason": null,"deletion_date": null,"deletion_by": null,"create_date": "2025-09-15T16:56:58","creator_by": "231b2c7d-a2eb-47db-a40d-d400f7d250f6","last_updated_date": null,"last_updated_by": null,"id": "3822eb35-ecc7-494a-b992-344523c1ead2"}]},"faulty": {"count": 0,"details": []}}
}初始化信息表
env
修改config\.env文件
# Message
#默認為中文
MESSAGE_DEFAULT_LANGUAGE=zh
#語言種類
MESSAGE_LANGUAGES=zh,en,fr,zh-TW
MESSAGE_FILE_PATH=config\messages_ini.jsonload_json
新建src\core\load_json.py文件
# src/core/json_loader.py
import json
from pathlib import Path
from typing import Any, Dictdef load_json(file_name: str, *, encoding: str = "utf-8") -> Dict[str, Any]:"""公共函数：读取项目根目录下任意 JSON 文件并返回 dict参数----file_name : str相对于项目根目录的完整路径，例如 "conf/message.json"encoding : str文件编码，默认 utf-8返回----Dict[str, Any]反序列化后的 JSON 对象"""# 1. 定位项目根目录（与 manage.py / main.py 同级）# root = Path(__file__).parent # .resolve().parent.parent.parent# 获取当前文件的绝对路径Root = Path(__file__).resolve().parent.parent.parent file_path = Root / file_name # 2. 读取并返回with file_path.open("r", encoding=encoding) as f:return json.load(f)
config
from src.core import json_loader # Message 配置MESSAGE_DEFAULT_LANGUAGE = os.getenv('MESSAGE_DEFAULT_LANGUAGE')# 读取环境变量时给默认值，避免空值导致列表异常MESSAGE_LANGUAGES = os.getenv('MESSAGE_LANGUAGES', 'zh,en,fr,zh-TW').split(',')MESSAGE_FILE_PATH=os.getenv('MESSAGE_FILE_PATH', '')MESSAGE_CONTENT_JSON=json_loader.load_json(MESSAGE_FILE_PATH)
json
新建config\messages_ini.json文件
{"messages": {"zh": {"9000": {"message": "%s操作成功", "desc": "操作成功"},"9001": {"message": "%s新增成功", "desc": "新增成功"},"9002": {"message": "%s修改成功", "desc": "修改成功"},"9003": {"message": "%s删除成功", "desc": "删除成功"},"9100": {"message": "%s操作失败", "desc": "通用操作异常"},"9101": {"message": "%s新增失败%s", "desc": "新增异常，带原因"},"9102": {"message": "%s修改失败%s", "desc": "修改异常，带原因"},"9103": {"message": "%s删除失败%s", "desc": "删除异常，带原因"},"9200": {"message": "%s已存在", "desc": "唯一键冲突"},"9201": {"message": "%s不存在", "desc": "主键/外键不存在"},"9202": {"message": "%s状态不正确", "desc": "状态机校验失败"},"9203": {"message": "%s已被锁定", "desc": "并发锁定提示"},"9210": {"message": "%s为空，无法提取主键值", "desc": "唯一键冲突"},"9211": {"message": "主键值列表为空或包含 None", "desc": "唯一键冲突"},"9212": {"message": "对象:%s 内容为空", "desc": "唯一键冲突"},"9300": {"message": "不是有效API Key:%s", "desc": "身份校验失败"},"9301": {"message": "API Key已停用", "desc": "身份校验失败"},"9302": {"message": "API Key已过期", "desc": "身份校验失败"},"9303": {"message": "API Key已删除", "desc": "身份校验失败"},"9321": {"message": "无权限操作%s", "desc": "功能级权限不足"},"9322": {"message": "账号%s已被禁用", "desc": "账号状态异常"},"9400": {"message": "参数%s不能为空", "desc": "必填校验"},"9401": {"message": "参数%s格式错误", "desc": "正则/格式校验"},"9402": {"message": "参数%s超出范围", "desc": "数值范围校验"},"9403": {"message": "参数%s长度必须在%s到%s 之间", "desc": "长度区间校验"},"9500": {"message": "系统繁忙，请稍后再试", "desc": "兜底系统异常"},"9501": {"message": "调用%s服务超时", "desc": "第三方超时"},"9502": {"message": "调用%s服务返回错误：%s", "desc": "第三方业务错误"},"9600": {"message": "共计:%s笔资料", "desc": "资料笔数"},"9601": {"message": "无任何资料", "desc": "资料笔数"},"9610": {"message": "成功创建%s笔资料", "desc": "资料笔数"},"9611": {"message": "资料已存在：跳过%s笔资料", "desc": "资料笔数"},"9612": {"message": "有%s个数据格式错误", "desc": "资料笔数"},"9630": {"message": "成功删除%s笔资料", "desc": "资料笔数"},"9631": {"message": "成功标注删除%s笔资料", "desc": "资料笔数"}},"en": {"9000": {"message": "%s operation successful", "desc": "Operation success"},"9001": {"message": "%s added successfully", "desc": "Add success"},"9002": {"message": "%s modified successfully", "desc": "Update success"},"9003": {"message": "%s deleted successfully", "desc": "Delete success"},"9100": {"message": "%s operation failed", "desc": "Generic operation error"},"9101": {"message": "%s add failed: %s", "desc": "Add error with reason"},"9102": {"message": "%s update failed: %s", "desc": "Update error with reason"},"9103": {"message": "%s delete failed: %s", "desc": "Delete error with reason"},"9200": {"message": "%s already exists", "desc": "Unique key conflict"},"9201": {"message": "%s does not exist", "desc": "Primary/foreign key not found"},"9202": {"message": "%s status incorrect", "desc": "State machine validation failed"},"9203": {"message": "%s is locked", "desc": "Concurrent lock prompt"},"9210": {"message": "%s is empty, unable to extract primary key", "desc": "Unique key conflict"},"9211": {"message": "Primary key list is empty or contains None", "desc": "Unique key conflict"},"9212": {"message": "Object:%s content is empty", "desc": "Unique key conflict"},"9300": {"message": "Invalid API Key: %s", "desc": "Authentication failed"},"9301": {"message": "API Key is disabled", "desc": "Authentication failed"},"9302": {"message": "API Key has expired", "desc": "Authentication failed"},"9303": {"message": "API Key has been deleted", "desc": "Authentication failed"},"9321": {"message": "No permission to operate %s", "desc": "Insufficient function-level permission"},"9322": {"message": "Account %s has been disabled", "desc": "Account status exception"},"9400": {"message": "Parameter %s cannot be empty", "desc": "Required validation"},"9401": {"message": "Parameter %s format error", "desc": "Regex/format validation"},"9402": {"message": "Parameter %s out of range", "desc": "Numeric range validation"},"9403": {"message": "Parameter %s length must be between %s and %s", "desc": "Length range validation"},"9500": {"message": "System busy, please try again later", "desc": "Fallback system exception"},"9501": {"message": "Call %s service timeout", "desc": "Third-party timeout"},"9502": {"message": "Call %s service returned error: %s", "desc": "Third-party business error"},"9600": {"message": "Total: %s records", "desc": "Record count"},"9601": {"message": "No records found", "desc": "Record count"},"9610": {"message": "Successfully created %s records", "desc": "Record count"},"9611": {"message": "Records already exist: skipped %s records", "desc": "Record count"},"9612": {"message": "%s data format errors", "desc": "Record count"},"9630": {"message": "Successfully deleted %s records", "desc": "Record count"},"9631": {"message": "Successfully marked deleted %s records", "desc": "Record count"}},"fr": {"9000": {"message": "Opération %s réussie", "desc": "Succès de l'opération"},"9001": {"message": "Ajout %s réussi", "desc": "Succès de l'ajout"},"9002": {"message": "Modification %s réussie", "desc": "Succès de la modification"},"9003": {"message": "Suppression %s réussie", "desc": "Succès de la suppression"},"9100": {"message": "Opération %s échouée", "desc": "Erreur d'opération générique"},"9101": {"message": "Ajout %s échoué : %s", "desc": "Erreur d'ajout avec raison"},"9102": {"message": "Modification %s échouée : %s", "desc": "Erreur de modification avec raison"},"9103": {"message": "Suppression %s échouée : %s", "desc": "Erreur de suppression avec raison"},"9200": {"message": "%s existe déjà", "desc": "Conflit de clé unique"},"9201": {"message": "%s n'existe pas", "desc": "Clé primaire/étrangère introuvable"},"9202": {"message": "Statut %s incorrect", "desc": "Échec de validation de la machine d'état"},"9203": {"message": "%s est verrouillé", "desc": "Verrouillage concurrent"},"9210": {"message": "%s est vide, impossible d'extraire la clé primaire", "desc": "Conflit de clé unique"},"9211": {"message": "La liste des clés primaires est vide ou contient None", "desc": "Conflit de clé unique"},"9212": {"message": "Objet:%s contenu vide", "desc": "Conflit de clé unique"},"9300": {"message": "Clé API invalide : %s", "desc": "Échec de l'authentification"},"9301": {"message": "Clé API désactivée", "desc": "Échec de l'authentification"},"9302": {"message": "Clé API expirée", "desc": "Échec de l'authentification"},"9303": {"message": "Clé API supprimée", "desc": "Échec de l'authentification"},"9321": {"message": "Pas de permission pour opérer %s", "desc": "Permission insuffisante au niveau fonction"},"9322": {"message": "Le compte %s a été désactivé", "desc": "Exception de statut de compte"},"9400": {"message": "Le paramètre %s ne peut pas être vide", "desc": "Validation obligatoire"},"9401": {"message": "Erreur de format du paramètre %s", "desc": "Validation de format/regex"},"9402": {"message": "Le paramètre %s est hors limite", "desc": "Validation de plage numérique"},"9403": {"message": "La longueur du paramètre %s doit être entre %s et %s", "desc": "Validation de plage de longueur"},"9500": {"message": "Système occupé, veuillez réessayer plus tard", "desc": "Exception système de secours"},"9501": {"message": "Appel du service %s expiré", "desc": "Expiration tiers"},"9502": {"message": "L'appel du service %s a renvoyé une erreur : %s", "desc": "Erreur métier tiers"},"9600": {"message": "Total : %s enregistrements", "desc": "Nombre d'enregistrements"},"9601": {"message": "Aucun enregistrement trouvé", "desc": "Nombre d'enregistrements"},"9610": {"message": "Création réussie de %s enregistrements ", "desc": "Nombre d'enregistrements"},"9611": {"message": "Enregistrements déjà existants : %s enregistrements ignorés", "desc": "Nombre d'enregistrements"},"9612": {"message": "%s erreurs de format de données", "desc": "Nombre d'enregistrements"},"9630": {"message": "Suppression réussie de %s enregistrements", "desc": "Nombre d'enregistrements"},"9631": {"message": "Marquage de suppression réussi de %s enregistrements", "desc": "Nombre d'enregistrements"}},"zh-TW": {"9000": {"message": "%s操作成功", "desc": "操作成功"},"9001": {"message": "%s新增成功", "desc": "新增成功"},"9002": {"message": "%s修改成功", "desc": "修改成功"},"9003": {"message": "%s刪除成功", "desc": "刪除成功"},"9100": {"message": "%s操作失敗", "desc": "通用操作異常"},"9101": {"message": "%s新增失敗%s", "desc": "新增異常，帶原因"},"9102": {"message": "%s修改失敗%s", "desc": "修改異常，帶原因"},"9103": {"message": "%s刪除失敗%s", "desc": "刪除異常，帶原因"},"9200": {"message": "%s已存在", "desc": "唯一鍵衝突"},"9201": {"message": "%s不存在", "desc": "主鍵/外鍵不存在"},"9202": {"message": "%s狀態不正確", "desc": "狀態機校驗失敗"},"9203": {"message": "%s已被鎖定", "desc": "併發鎖定提示"},"9210": {"message": "%s為空，無法提取主鍵值", "desc": "唯一鍵衝突"},"9211": {"message": "主鍵值列表為空或包含 None", "desc": "唯一鍵衝突"},"9212": {"message": "物件:%s 內容為空", "desc": "唯一鍵衝突"},"9300": {"message": "不是有效API Key:%s", "desc": "身份校驗失敗"},"9301": {"message": "API Key已停用", "desc": "身份校驗失敗"},"9302": {"message": "API Key已過期", "desc": "身份校驗失敗"},"9303": {"message": "API Key已刪除", "desc": "身份校驗失敗"},"9321": {"message": "無權限操作%s", "desc": "功能級權限不足"},"9322": {"message": "帳號%s已被禁用", "desc": "帳號狀態異常"},"9400": {"message": "參數%s不能為空", "desc": "必填校驗"},"9401": {"message": "參數%s格式錯誤", "desc": "正則/格式校驗"},"9402": {"message": "參數%s超出範圍", "desc": "數值範圍校驗"},"9403": {"message": "參數%s長度必須在%s到%s 之間", "desc": "長度區間校驗"},"9500": {"message": "系統繁忙，請稍後再試", "desc": "兜底系統異常"},"9501": {"message": "呼叫%s服務超時", "desc": "第三方超時"},"9502": {"message": "呼叫%s服務返回錯誤：%s", "desc": "第三方業務錯誤"},"9600": {"message": "共計:%s筆資料", "desc": "資料筆數"},"9601": {"message": "無任何資料", "desc": "資料筆數"},"9610": {"message": "成功建立%s筆資料", "desc": "資料筆數"},"9611": {"message": "資料已存在：跳過%s筆資料", "desc": "資料筆數"},"9612": {"message": "有%s個資料格式錯誤", "desc": "資料筆數"},"9630": {"message": "成功刪除%s筆資料", "desc": "資料筆數"},"9631": {"message": "成功標註刪除%s筆資料", "desc": "資料筆數"}}},"messagesNoAPI": {"zh": {"9390": {"message": "传输Header无Token资料", "desc": "身份校验失败"},"9391": {"message": "传输Header无API Key资料", "desc": "身份校验失败"},"9392": {"message": "Token失效或者已过期", "desc": "身份校验失败"},"9393": {"message": "Token解析API Key与Header中API KEY不一致", "desc": "身份校验失败"},"9394": {"message": "Token payload失败%s", "desc": "身份校验失败"}},"en": {"9390": {"message": "Missing Token in Header", "desc": "Authentication failed"},"9391": {"message": "Missing API Key in Header", "desc": "Authentication failed"},"9392": {"message": "Invalid or expired JWT Token", "desc": "Authentication failed"},"9393": {"message": "The token API key does not match the login API key.", "desc": "Authentication failed"},"9394": {"message": "Token payload invalid:%s", "desc": "Authentication failed"}},"fr": {"9390": {"message": "Token manquant dans l'en-tête", "desc": "Échec de l'authentification"},"9391": {"message": "Clé API manquante dans l'en-tête", "desc": "Échec de l'authentification"},"9392": {"message": "Token invalide ou expiré", "desc": "Échec de l'authentification"},"9393": {"message": "La clé API du token ne correspond pas à celle de la connexion.", "desc": "Échec de l'authentification"},"9394": {"message": "Échec du payload du token : %s", "desc": "Échec de l'authentification"}},"zh-TW": {"9390": {"message": "傳輸Header無Token資料", "desc": "身份校驗失敗"},"9391": {"message": "傳輸Header無API Key資料", "desc": "身份校驗失敗"},"9392": {"message": "Token失效或者已過期", "desc": "身份校驗失敗"},"9393": {"message": "Token解析API Key與Header中API KEY不一致", "desc": "身份校驗失敗"},"9394": {"message": "Token payload失敗%s", "desc": "身份校驗失敗"}}}
}
message初始化
修改src/crud/sys_message.py文件
def ini_msg_exists(db: Session, api_id: str, language: str, code: int) -> bool:return db.query(models.SysMessage).filter(models.SysMessage.api_id == api_id,models.SysMessage.language == language,models.SysMessage.code == code,models.SysMessage.deletion_mark == 0).first() is not Nonedef ini_create_message(db: Session, project: schemas.SysMessageCreate):db_message = models.SysMessage(**project.dict())db.add(db_message)db.commit()db.refresh(db_message)return db_messagedef ini_message(db: Session):"""初始化语言消息。"""# 1. 按语言分层LANGUAGES = Config.MESSAGE_LANGUAGESMESSAGE_CONTENT_JSON=Config.MESSAGE_CONTENT_JSONmessages = MESSAGE_CONTENT_JSON.get('messages', {})messagesNoAPI = MESSAGE_CONTENT_JSON.get('messagesNoAPI', {}) # json.loads(Config.MESSAGE_NO_API_KEY_CONTENT)sysApiProjects = SysApiProjectCRUD.ini_get_api_project_by_api_code(db, None)for sysApiProject in sysApiProjects:api_id = sysApiProject.idfor lang in LANGUAGES: # ["zh", "en", "fr", "zh-TW"]:lang_dict = messages.get(lang, {})for code, item in lang_dict.items():msg = item["message"]desc = item["desc"]if not ini_msg_exists(db, api_id, lang, code):ini_create_message(db, schemas.SysMessageCreate(api_id=api_id,language=lang,code=code,message=msg,description=desc))api_id = "-1"for lang in LANGUAGES: # ["zh", "en", "fr", "zh-TW"]:lang_dict = messagesNoAPI.get(lang, {})for code, item in lang_dict.items():msg = item["message"]desc = item["desc"]if not ini_msg_exists(db, api_id, lang, code):ini_create_message(db, schemas.SysMessageCreate(api_id=api_id,language=lang,code=code,message=msg,description=desc))
功能性说明：
ini_message() 是“国际化消息一键初始化”脚本，仅在系统首次部署或新增语言包时运行。它读取配置文件中的 MESSAGE_LANGUAGES 与 MESSAGE_CONTENT_JSON ，自动完成以下任务：
1. 去重校验
ini_msg_exists() 按 (api_id, language, code, deletion_mark=0) 查询，已存在则跳过，防止重复插入。
2. 项目级消息写入
遍历 sys_api_project 全表，取得每个 API 的 id 作为 api_id ，再逐语言、逐错误码落地 messages 段内容；实现“一个接口一套多语提示”的细粒度管理。
3. 全局通用消息写入
当 api_id = "-1" 时，写入 messagesNoAPI 段，专门存放登录、网关、全局异常等无特定接口的公共文案，保证所有提示都有多语版本。
4. 原子提交
每条新增走独立事务 ini_create_message() ，失败自动回滚，日志定位到具体 (api_id, lang, code) ，便于排错。
通过该初始化函数，运维人员只需维护 JSON 配置即可在秒级完成全量多语消息的录入与更新，无需手工 SQL，实现“配置即文案”的自动化国际化交付。
项目初始化
修改crud/sys_api_project.py
from src import logger_cust
# 初始化资料源
def ini_get_api_project_by_api_code(db: Session, api_code: str | None):query = db.query(models.SysApiProject)if api_code is not None: # 非空才过滤return query.filter(models.SysApiProject.api_code == api_code).first() is not Nonereturn query.all() # 空串/None 返回全部def ini_create_api_project(db: Session, project: schemas.SysApiProjectCreate):db_project = models.SysApiProject(**project.dict())db.add(db_project)db.commit()db.refresh(db_project)return db_projectdef ini_api_project(db: Session):api_project = ini_get_api_project_by_api_code(db, "api")if not api_project:ini_create_api_project(db, schemas.SysApiProjectCreate(name="API",api_code="api",api_secret="api",api_url="",api_status=True,))logger_cust.debug("sys_mangers-init_project:默认项目 'apiAdmin' 已添加到 sys_api_project 表中。")else:logger_cust.info("sys_mangers-init_project:默认项目 'apiAdmin' 已存在，无需操作。")
功能性说明
ini_api_project() 负责“默认资料源”的自动植入，仅在系统首次启动或容器重建时执行，逻辑如下：
1. 查询判定
ini_get_api_project_by_api_code() 根据传入的 api_code 进行精准匹配；若传入 None 则返回整张表，用于后续批量初始化。
2. 防重复写入
当检测到 api_code = "api" 的记录已存在时直接跳过，避免重复主键冲突，并记录 INFO 日志提示。
3. 默认资料源创建
未发现记录时，自动写入一条内置项目：名称 "API" 、密钥 "api" 、状态启用，作为后续国际化消息、网关路由、调用链追踪的“根节点”资料源。
4. 原子提交与日志
单事务提交，失败自动回滚；DEBUG/INFO 级别日志明确记录“已添加”或“已存在”，方便运维追踪初始化过程。
通过该函数，系统无需人工预置数据即可拥有一个可用的默认 API 项目，为后续消息绑定、权限划分、流量统计提供统一的“资料源”基准。
项目初始化
修改\src\core\sys_mangers.py文件
# \src\core\sys_mangers.py
from src.crud import sys_api_project as SysApiProjectCrud,sys_message as SysMessageCrud
from src.schemas.sys_api_project import SysApiProjectCreate
from sqlalchemy.orm import Session
from src import logger_custdef init_project(db: Session):SysApiProjectCrud.ini_api_project(db)SysMessageCrud.ini_message(db)功能性说明
sys_managers.py 是“系统级初始化编排器”，仅暴露一个入口函数 init_project(db: Session) ，职责如下：
1. 一键顺序执行
先调用 SysApiProjectCrud.ini_api_project() 确保默认资料源存在，再调用 SysMessageCrud.ini_message() 写入国际化消息；两步串行，后者依赖前者生成的 api_id ，保证外键逻辑正确。
2. 幂等安全
内部所有 CRUD 均自带“存在即跳过”机制，可重复执行，适用于容器重启、CI 升级、多实例并发部署场景，不会产生脏数据或主键冲突。
3. 集中日志
统一使用 logger_cust 输出，初始化结果一目了然，方便接入运维监控或 K8s Job 状态检查。
通过该模块，开发/运维人员只需在应用启动时调用一次 init_project() ，即可完成“资料源 + 多语消息”双层初始化，实现“零手工 SQL、零配置遗漏”的自动化交付。
运行
终止项目重新运行，在数据库中查询表(sys_message),资料是否被初始化。
封装
信息封装
修改src\core\retMes.py
# src\core\retMes.py
from fastapi import HTTPException
from src.models.sys_message import SysMessage
from src.core import str_utils as strU
from src.core.config import Config
# 在原文件末尾追加（render_msg 与 I18nMsg 放一起）
def render_msg(db, api_id, code, language=None, *args):"""从 sys_message 取模板并填充 %s"""if strU.is_empty(language):language = Config.MESSAGE_DEFAULT_LANGUAGErow = db.query(SysMessage.message).filter(SysMessage.api_id == api_id,SysMessage.language == language,SysMessage.code == code,SysMessage.deletion_mark == 0).first()if not row:return f"Unknown error {code}"tpl = row.messagetpl = row.messageneed_cnt = tpl.count("%s")use_args = args[:need_cnt]print(use_args)print(tpl)retMessage=tpl % use_args if use_args else tplretMessage = retMessage.replace("：%s", "")retMessage = retMessage.replace(":%s", "")retMessage=retMessage.replace("%s", "")return retMessageclass Msg:def __init__(self, db, api_id, code, language="zh", *args):self.msg = render_msg(db, api_id,code, language, *args)def mes(self):return self.msg
功能性说明
render_msg() 与 Msg 组成“运行时国际化消息渲染器”，在任意业务代码中提供“一行代码获取多语提示”的能力：
1. 模板定位
根据 (api_id, code, language) 三键查询 sys_message 表，已逻辑删除或缺失返回 "Unknown error {code}" ，防止前端空指针。
2. 占位符填充
支持可变参数 *args ，按 %s 数量自动截取，避免参数过多或不足；填充后二次清洗，去掉未替换的“：%s”“:%s”“%s”残串，保证输出干净、可读。
3. 缺省语言
当调用方未传 language 时，自动取配置项 MESSAGE_DEFAULT_LANGUAGE ，保证后台任务、异步线程也能拿到稳定的语言版本。
4. 便捷封装
Msg 类把上述过程封装成对象，用法：
Msg(db, api_id, 1001, "en", userName).mes() → "User john exceeds limit."
业务层无需关心查询与格式化细节，符合“一处定义、处处复用”的国际化设计原则。
调用调整
修改src\core\orm_curd.py文件
# src\core\orm_curd.py
from typing import Set, Any, Dict, List, Optional, Type, TypeVar, Tuple, Unionfrom pyexpat.errors import messages
from sqlalchemy.orm import Session, Query
from pydantic import BaseModel
from src.core import retMes, str_utils as strU
from sqlalchemy import and_, or_
from sqlalchemy.inspection import inspect
from src.schemas.request_model import UpdateModel, UpdateItem, ensure_dict, DeleteModel, DeleteItem
from src.schemas.login_manager import AuthManagerT = TypeVar("T") # SQLAlchemy Model
S = TypeVar("S", bound=BaseModel) # Pydantic Schemadef _multi_filter(Model: Type[T], unique_fields: List[str], row: Dict[str, Any]):"""根据多个字段生成 and_ 条件"""return and_(*(getattr(Model, k) == row[k] for k in unique_fields))def Insert(db: Session,auth: AuthManager,Model: Type[T],Schema: Type[S],Model_Name: str,data: List[Dict[str, Any]],unique_fields: List[str],skip_duplicates: bool = True,
) -> Dict[str, Any]: # Tuple[List[T], List[T]]:"""通用批量写入/跳过重复:param db: SQLAlchemy Session:param auth: 登录信息:param Model: 数据库模型类:param Schema: Pydantic 校验模型（仅用于序列化返回）:param Model_Name:数据库模型类名称:param data: 待写入的字典列表:param unique_fields: 用于判断是否重复的唯一字段名列表:param skip_duplicates: True=跳过重复；False=抛异常:return: 按 retMes.Success 包装好的 dict"""api_id = auth.api_iduser_id = auth.user_idlanguage=auth.languagecreated, duplicates, faulty = [], [], []for row in data:if not isinstance(row, dict):row = row.model_dump() #V1.dict(),V2 model_dump() # 不是 dict 转换dict结构# --------- 新增：创建人、更新人等默认值 ---------if hasattr(Model, 'api_id'):row["api_id"] = api_idif hasattr(Model, 'api_id'):row["creator_by"] = user_idif hasattr(Model, 'last_updated_date'):row["last_updated_date"] = Noneif hasattr(Model, 'last_updated_by'):row["last_updated_by"] = Noneif hasattr(Model, 'deletion_date'):row["deletion_date"] = Noneif hasattr(Model, 'deletion_by'):row["deletion_by"] = None# --------- 新增：逻辑删除 ---------if hasattr(Model, 'deletion_mark'):if row["deletion_mark"]:if hasattr(Model, 'deletion_date') and row["deletion_date"] is None:row["deletion_date"] = strU.now()if hasattr(Model, 'deletion_by'):row["deletion_by"] = user_id# --------- 新增：完整性校验 ---------missing = [k for k in unique_fields if k not in row or row[k] is None]if missing:faultyData = {"message": f"行数据缺少唯一键字段为：{missing}","details": row}faulty.append(faultyData)continue# 1. 构造唯一性过滤条件exists = db.query(Model).filter(_multi_filter(Model, unique_fields, row)).first()# filter_cond = [getattr(Model, k) == row[k] for k in unique_fields]# exists = db.query(Model).filter(*filter_cond).first()if exists:duplicates.append(exists)if not skip_duplicates:retMes.Raise(message=f"Duplicate on {unique_fields}: {row}").mes()# raise ValueError(f"Duplicate on {unique_fields}: {row}")continuenew_obj = Model(**row)db.add(new_obj)created.append(new_obj)if created:db.commit()for obj in created:db.refresh(obj)# 2. 统一序列化# created_dicts = [strU.model_to_dict(obj) for obj in created]# duplicates_dicts = [Schema.from_orm(obj).dict() for obj in duplicates]created_dicts = [Schema.model_validate(obj).model_dump() for obj in created]duplicates_dicts = [Schema.model_validate(obj).model_dump() for obj in duplicates]# 3. 构造响应message=Noneif created_dicts:message=retMes.Msg(db, api_id, '9610', language, len(created_dicts)).mes()if duplicates_dicts:msg=retMes.Msg(db, api_id, '9611', language, len(duplicates_dicts)).mes()if message:message += f"， {msg}"else:message = msgif faulty:msg=retMes.Msg(db, api_id, '9612', language, len(faulty)).mes()if message:message += f"， {msg}"else:message = msgresponse_data = {"created": {"count": len(created_dicts), "details": created_dicts},"duplicates": {"count": len(duplicates_dicts), "details": duplicates_dicts},"faulty": {"count": len(faulty), "details": faulty},}return retMes.Success(response_data, message).mes()def query_with_page(db: Session,auth: AuthManager,model_cls: Type, # 对应 models.SysUserresp_model_cls: Optional[Type] = None, # 对应 schemas.SysUser，仅做字段校验，可省略*,id: Optional[str] = None,filter_model: Optional[BaseModel] = None, # 对应 request_model.QueryModelpage: int = 0,page_size: int = 100
) -> Dict[str, Any]:"""通用分页查询:param db: 数据库会话:param auth: 登录信息:param model_cls: SQLAlchemy 实体类:param resp_model_cls: Pydantic 响应模型（可选）:param id: 按主键 id 精确查询:param filter_model: 前端传来的过滤 + 分页参数:param page: 默认页码:param page_size: 默认每页条数:return: 按 retMes.Success 包装好的 dict"""api_id = auth.api_idlanguage = auth.language# query: Query = db.query(model_cls).filter(model_cls.api_id == api_id)if hasattr(model_cls, 'api_id'):query: Query = db.query(model_cls).filter(model_cls.api_id == api_id)else:query: Query = db.query(model_cls) # 不带 api_id 的表直接查全量# 1. 主键精确查询if strU.is_not_empty(id):query = query.filter(model_cls.id == id)# 2. 动态过滤条件elif strU.is_not_empty(filter_model):if filter_model.page is not None:page = filter_model.pageif filter_model.page_size is not None:page_size = filter_model.page_sizefilter_data: Dict[str, Any] = filter_model.data or {}for key, value in filter_data.items():if hasattr(model_cls, key) and value is not None:query = query.filter(getattr(model_cls, key) == value)# 3. 分页def _calc_skip(pg: int, ps: int) -> int:return (pg - 1) * ps + 1 if pg > 1 else 0if strU.is_empty(page) or strU.is_empty(page_size):total = query.count()rows = query.all()else:skip = _calc_skip(page, page_size)limit = page_sizetotal = query.count()rows = query.offset(skip).limit(limit).all()# 4. 转 dict# details: List[Dict[str, Any]] = [strU.model_to_dict(r) for r in rows]# 4. 转 dict（带 resp_model_cls 校验）details: List[Dict[str, Any]] = []for row in rows:if resp_model_cls: # ← 新增validated = resp_model_cls.from_orm(row) # Pydantic 校验/脱敏details.append(validated.dict())else: # ← 原有details.append(strU.model_to_dict(row))# 5. 包装payload = {"count": len(details),"page": page,"page_size": page_size,"total_count": total,"details": details}message = retMes.Msg(db, api_id, '9600', language, len(details)).mes() #f'共计：{len(details)} 条数据'return retMes.Success(payload,message,record_count=len(details)).mes()class PkValuesEmptyError(ValueError):"""主键值列表为空异常"""passdef _only_schema_fields(obj: Any, schema: Union[BaseModel, Type[BaseModel], dict]) -> Dict[str, Any]:"""只保留 schema 中声明的字段（支持 Pydantic 模型或 dict）"""if isinstance(schema, dict):schema_fields = schema.keys()elif isinstance(schema, BaseModel):schema_fields = schema.__fields_set__ | schema.__fields__.keys()else: # 类类型schema_fields = schema.__fields__.keys()# 2. 统一把 obj 变 dictif isinstance(obj, dict):data = objelse:data = strU.model_to_dict(obj) # 仅 ORM 对象才需要# 3. 过滤return {k: v for k, v in data.items() if k in schema_fields}def updateItem(db: Session,auth: AuthManager,model_cls,pk_names: List[str],pk_values: Optional[List[Any]] = None,schema_obj: Union[BaseModel, Dict[str, Any]] = None,update_type: str = "update",
) -> Optional[Any]:"""通用实体部分更新（支持联合主键）。如果调用方未传入 pk_values，则尝试从 schema_obj 解析；解析后仍为空则抛出 PkValuesEmptyError。:param db: 数据库会话:param auth: 登录信息:param model_cls: ORM 模型类:param pk_names: 主键字段名列表，如 ['tenant_id', 'id']:param pk_values: 主键值列表，如 [1, 'abc']；可为 None:param schema_obj: Pydantic 实例（含待更新字段）:param update_type: 修改类型:return: 更新后的 ORM 对象或 None（未找到）:raises: PkValuesEmptyError"""# print(schema_obj)api_id = auth.api_iduser_id = auth.user_idlanguage = auth.language# 1. 若未显式传入 pk_values，则从 schema_obj 提取if strU.is_empty(pk_values):if schema_obj is None:message = retMes.Msg(db, api_id, '9210', language, 'schema_obj').mes()retMes.Raise(message=message).mes()# raise PkValuesEmptyError("schema_obj 为空，无法提取主键值")# data = schema_obj.dict()data = ensure_dict(schema_obj)pk_values = [data.get(name) for name in pk_names]# 2. 再次检查 pk_values 是否全部非空if not pk_values or any(v is None for v in pk_values):message = retMes.Msg(db, api_id, '9211', language).mes()retMes.Raise(message=message).mes()# raise PkValuesEmptyError("主键值列表为空或包含 None")# 3. 构造查询并更新# query = db.query(model_cls)if hasattr(model_cls, 'api_id'):query = db.query(model_cls).filter(model_cls.api_id == api_id)else:query = db.query(model_cls) # 不带 api_id 的表直接查全量for name, value in zip(pk_names, pk_values):query = query.filter(getattr(model_cls, name) == value)# 如果update_type=“mark”，增加deletion_mark为0或者false判断if update_type == "mark":query = query.filter(model_cls.deletion_mark == 0)# 4. 先捞出所有主键（仅主键，节省内存）----------# 支持联合主键：每条主键是一个 tuplemapper = inspect(model_cls)pk_cols = mapper.primary_keyif pk_cols is None:pk_cols = [getattr(model_cls, pk) for pk in pk_names]pk_list = query.with_entities(*pk_cols).all() # [(val1, val2), ...]if not pk_list:return []# 5. 组装更新字段update_data = ensure_dict(schema_obj)update_data = {k: v for k, v in update_data.items()if k not in pk_names and k not in {"create_date", "creator_by"}}if hasattr(model_cls, "last_updated_date"):update_data["last_updated_date"] = strU.now()if hasattr(model_cls, "last_updated_by"):update_data["last_updated_by"] = user_idif not update_data: # 无字段可更新return []# 6. 批量更新query.update(update_data, synchronize_session=False)db.commit()# 7. 根据主键再查一次，拿到更新后实体 ----------# 动态构造 OR (pk1=val1 AND pk2=val2) ...or_conditions = [and_(*[col == val for col, val in zip(pk_cols, vals)])for vals in pk_list]updated_rows = db.query(model_cls).filter(or_(*or_conditions)).all()# 8. 按 schema_obj 字段过滤if schema_obj is None:return [strU.model_to_dict(r) for r in updated_rows]return [_only_schema_fields(r, schema_obj) for r in updated_rows]def update(db: Session,auth: AuthManager,model_cls,update_obj: Optional[UpdateModel] = None,
) -> Dict[str, Any]:"""通用实体部分更新（支持联合主键）。如果调用方未传入 pk_values，则尝试从 schema_obj 解析；解析后仍为空则抛出 PkValuesEmptyError。:param db: 数据库会话:param auth: 登录信息:param model_cls: ORM 模型类:param update_obj: Pydantic 实例（含待更新字段）:return: 更新后的 ORM 对象或 None（未找到）:raises: PkValuesEmptyError"""# 1.api_id = auth.api_idlanguage = auth.languageif update_obj is None:message = retMes.Msg(db, api_id, '9212', language, 'update_obj').mes()return retMes.Error(message).mes()updated_entities: List[Any] = []# 遍历裸数组for item in update_obj.root:# 如果 item 里给了 pk_names / pk_values，就用它；否则用外层默认值_pk_names = item.pk_names if item.pk_names is not None else None_pk_values = item.pk_values if item.pk_values is not None else None_data = item.data if item.data is not None else Noneentity = updateItem(db=db,auth=auth,model_cls=model_cls,pk_names=_pk_names,pk_values=_pk_values,schema_obj=_data, # 把 data 字典作为 schema_obj)if entity is not None:updated_entities.extend(entity) # entity_dict# entity_dict = strU.model_to_dict(entity)# append(obj) 把 obj 原封不动当成一个元素追加到列表末尾。 结果长度 +1# extend(iterable) # 把 iterable 拆开成单个元素后再依次追加。 # 结果长度 +len(iterable)# updated_entities.append(entity_dict)message = retMes.Msg(db, api_id, '9600', language, len(updated_entities)).mes()return retMes.Success(updated_entities,message,record_count=len(updated_entities)).mes()def deleteItem(db: Session,auth: AuthManager,model_cls,pk_names: List[str],pk_values: Optional[List[Any]] = None,schema_obj: Union[BaseModel, Dict[str, Any]] = None,exclude_fields: Optional[Set[str]] = {"api_id", "creator_by"},
) -> List[Dict[str, Any]]:"""删除所有符合主键条件的记录（支持联合主键）返回实际删除的行数"""api_id = auth.api_idlanguage = auth.language# 1. 解析主键if strU.is_empty(pk_values):data = ensure_dict(schema_obj)pk_values = [data.get(name) for name in pk_names]if not pk_values or any(v is None for v in pk_values):message = retMes.Msg(db, api_id, '9211', language).mes()retMes.Raise(message=message).mes()# 2. 构造条件# query = db.query(model_cls)if hasattr(model_cls, 'api_id'):query: Query = db.query(model_cls).filter(model_cls.api_id == api_id)else:query: Query = db.query(model_cls) # 不带 api_id 的表直接查全量for name, value in zip(pk_names, pk_values):query = query.filter(getattr(model_cls, name) == value)# 3. 先取出字段（避免 detached 后访问失败）rows = query.all()if not rows:return []data_list = strU.model_to_dict(rows)# 3.2 统一剔除if exclude_fields:for item in data_list:for f in exclude_fields:item.pop(f, None)# 3. 批量删除（0 查询）row_count = query.delete(synchronize_session=False) # 直接发 DELETE ... WHERE ...db.commit()return data_listdef delete(db: Session,auth: AuthManager,model_cls,delete_obj: Optional[DeleteModel] = None,
) -> Dict[str, Any]:"""通用实体部分更新（支持联合主键）。如果调用方未传入 pk_values，则尝试从 schema_obj 解析；解析后仍为空则抛出 PkValuesEmptyError。:param db: 数据库会话:param auth: 登录信息:param model_cls: ORM 模型类:param delete_obj: Pydantic 实例（含待更新字段）:return: 更新后的 ORM 对象或 None（未找到）:raises: PkValuesEmptyError"""api_id = auth.api_idlanguage = auth.language# 1.if delete_obj is None:message = retMes.Msg(db, api_id, '9212', language, 'delete_obj').mes()return retMes.Error(message).mes()deleted_entities: List[Any] = []marked_entities: List[Any] = []# 遍历裸数组for item in delete_obj.root:# 如果 item 里给了 pk_names / pk_values，就用它；否则用外层默认值_deletion_model = item.deletion_model if item.deletion_model is not None else None_pk_names = item.pk_names if item.pk_names is not None else None_pk_values = item.pk_values if item.pk_values is not None else None_data = item.data if item.data is not None else None_deletion_reason = item.deletion_reason if item.deletion_reason is not None else Noneif _deletion_model == "mark":#增加Data内容，若没有，则增加一个if _data is None:if _pk_names is None:message = retMes.Msg(db, api_id, '9211', language).mes()retMes.Raise(message=message).mes()_data = {"deletion_reason": _deletion_reason}else:# 检查_data中是否有deletion_reason字段if "deletion_reason" not in _data:_data["deletion_reason"] = _deletion_reason# 增加deletion_date字段与强制增加删除日期_data["deletion_date"] = strU.now()_data["deletion_mark"] = 1# 标记删除marked = updateItem(db=db,auth=auth,model_cls=model_cls,pk_names=_pk_names,pk_values=_pk_values,schema_obj=_data, # 把 data 字典作为 schema_objupdate_type="mark")if marked is not None and len(marked) > 0:marked_entities.extend(marked)# print(marked)# marked_dict = strU.model_to_dict(marked)# marked_entities.extend(marked_dict)else:deleted = deleteItem(db=db,auth=auth,model_cls=model_cls,pk_names=_pk_names,pk_values=_pk_values,schema_obj=_data, # 把 data 字典作为 schema_obj)if len(deleted) > 0:deleted_entities.extend(deleted)"""if deleted is not None and len(deleted)>0:deleted_dict = strU.model_to_dict(deleted)deleted_entities.extend(deleted_dict)"""# 3. 构造响应message_parts = []if deleted_entities:message = retMes.Msg(db, api_id, '9630', language,len(deleted_entities)).mes()message_parts.append(message)if marked_entities:message = retMes.Msg(db, api_id, '9631', language, len(marked_entities)).mes()message_parts.append(message)message = retMes.Msg(db, api_id, '9601', language).mes()retMessage = "，".join(message_parts) or messageresponse_data = {"deleted": {"count": len(deleted_entities), "details": deleted_entities},"marked": {"count": len(marked_entities), "details": marked_entities},}record_count = len(deleted_entities) + len(marked_entities)return retMes.Success(response_data,f'{retMessage} ',record_count).mes()
案例：
在query_with_page调整原f'共计：{len(details)} 条数据'改为message = retMes.Msg(db, api_id, '9600', language, len(details)).mes()
其他文件调整如下：
修改src/core/dependencies.py
"""
验证APP Key和Token
"""
# src/core/dependencies.py
from fastapi import Depends, HTTPException, Header
from src.core.security import verify_token, get_app_id
from src.core import retMes
from fastapi import Request
from sqlalchemy.orm import Session, Query
from typing import Optional
from src.core.ormdb import get_db
from src.schemas.login_manager import AuthManager
from pydantic import ValidationErrordef auth_api_key(db: Session = Depends(get_db),api_key: str = Header(None, alias="x-api-key"),language: str = Header(None, alias="language")):# 为了接口命名规范，把API_KEY改为 x-api-key 2025/04/23# 根据 APP KEY 抓取项目配置APP ID request: Request# api_key = api_key api_key: str = Header(None, alias="API_KEY")# api_key: str = Header(...) 无法获取，通过 request.headers 获取"""headers = request.headersprint(headers)api_key = headers.get('api-key')if api_key is None:retMes.Raise(message='API_KEY header is missing').mes()"""app_id = get_app_id(db, api_key, language)# app_id = 1return {"app_id": app_id,"api_key": api_key,"language": language}def auth_token(db: Session = Depends(get_db),api_key: str = Header(None, alias="x-api-key"),token: str = Header(None, alias="token"),language: str = Header(None, alias="language")) -> AuthManager:# 为了接口命名规范，把API_KEY改为 x-api-key 2025/04/23payload = verify_token(db, api_key, token, language)if not payload:msg = retMes.Msg(db, '-1', 9392, language).mes()retMes.Raise(message=msg).mes()# raise HTTPException(status_code=401, detail="Invalid or expired JWT Token")if payload.get('api_id') != payload.get('app_id'):msg = retMes.Msg(db, '-1', 9393, language).mes()retMes.Raise(message=msg).mes()payload.pop('app_id', None)# return payloadtry:return AuthManager.model_validate(payload) # Pydantic v2# 如果是 v1 用 AuthManager.parse_obj(payload)except ValidationError as e:msg = retMes.Msg(db, '-1', 9394, language,e).mes()retMes.Raise(message=msg).mes()
修改src/core/security.py
"""
安全相关模块（如认证、授权等）。
"""""
from email.message import Message# src/core/security.py
import jwt
from datetime import datetime, timedelta
from src.core import Config, str_utils as stru, retMes
from src.models.sys_api_project import SysApiProject as sysApiProjectModels
from sqlalchemy.orm import Session, Query
from typing import Optionaldef get_app_id(db: Session,api_key: Optional[str],language: Optional[str] = None) -> str:"""根据传入的 API Key 查询对应的 app_id（项目编号）。功能说明--------1. 校验：API Key 不能为空（None 或空字符串均视为缺失）。2. 查询：在 sys_api_project 表（sysApiProjectModels）中按 api_key 精确匹配。3. 安全：未命中或 key 非法时抛出业务异常，由全局异常处理器统一返回 401。4. 返回：命中后只取 app_id 字段，不返回整行对象，减少序列化开销。参数----db : SessionSQLAlchemy 会话，由 FastAPI Depends 注入。api_key : Optional[str]请求头中的 API Key；允许 None 以便在网关层统一处理缺失场景。返回----str与 api_key 绑定的 app_id（业务方项目编号）。异常----ValueError由 retMes.Raise 抛出，全局异常处理器会转译为 401 Unauthorized。"""if stru.is_empty(api_key): # None、"" 都拦住msg = retMes.Msg(db, "-1", code="9390", language=language).mes()retMes.Raise(message=msg).mes()api_project = (db.query(sysApiProjectModels) # ← 这里传的是类，不是模块.filter(sysApiProjectModels.api_key == api_key).first())api_id = api_project.idif api_project is None:msg = retMes.Msg(db, api_id, "9300", language, api_key).mes()retMes.Raise(message=msg).mes()if api_project.is_activate == 0:msg = retMes.Msg(db, api_id, "9301", language).mes()retMes.Raise(message=msg).mes()if api_project.deletion_mark == 1:msg = retMes.Msg(db, api_id, code="9303", language=language).mes()retMes.Raise(message=msg).mes()if api_project.disable_date and api_project.disable_date < stru.now():msg = retMes.Msg(db, api_id, code="9302", language=language).mes()retMes.Raise(message=msg).mes()return api_iddef generate_token(data: dict,expires_delta: timedelta = None):to_encode = data.copy()if not expires_delta:expires_delta = timedelta(minutes=Config.ACCESS_TOKEN_EXPIRE_MINUTES)if expires_delta:expire = datetime.utcnow() + expires_deltaelse:expire = datetime.utcnow() + timedelta(minutes=15)to_encode.update({"exp": expire})encoded_jwt = jwt.encode(to_encode, Config.SECRET_KEY, algorithm="HS256")return encoded_jwtdef verify_token(db: Session,api_key: str,token: str,language: Optional[str] = None):if token is None:msg = retMes.Msg(db, "-1", code="9390", language=language).mes()retMes.Raise(message=msg).mes()app_id = get_app_id(db, api_key, language)try:payload = jwt.decode(token, Config.SECRET_KEY, algorithms=["HS256"])payload['app_id'] = app_idpayload['api_key'] = api_keypayload["language"] = language# print(stru.timestamp_to_date(payload['exp']))payload['exp'] = stru.timestamp_to_date(payload['exp'])return payloadexcept jwt.ExpiredSignatureError:return Noneexcept jwt.InvalidTokenError:return None
测试
1.Accept-Language为en
调整/users/案例，在Header增加Language值为en
运行结果："message": "Records already exist: skipped 3 records"
{"status_code": 200,"status": "success","message": "Records already exist: skipped 3 records","record_count": 0,"data": {"created": {"count": 0,"details": []},"duplicates": {"count": 3,"details": [{"login_id": "alice1736","user_name": "Alice","role": null,"email": "alice@example.com","person_id": "P001","is_superuser": false,"is_staff": true,"effective_date": "2023-01-01","expiry_date": "2025-12-31","deletion_mark": false,"deletion_reason": null,"deletion_date": null,"deletion_by": null,"create_date": "2025-09-15T16:56:58","creator_by": "231b2c7d-a2eb-47db-a40d-d400f7d250f6","last_updated_date": null,"last_updated_by": null,"id": "bb6b9a96-239b-4956-a2a3-c39d9f6113fe"},{"login_id": "bob4","user_name": "Bob","role": null,"email": "bob@example.com","person_id": "P002","is_superuser": false,"is_staff": false,"effective_date": "2023-02-01","expiry_date": "2025-12-31","deletion_mark": true,"deletion_reason": null,"deletion_date": "2025-09-15T16:59:29","deletion_by": "231b2c7d-a2eb-47db-a40d-d400f7d250f6","create_date": "2025-09-15T16:59:29","creator_by": "231b2c7d-a2eb-47db-a40d-d400f7d250f6","last_updated_date": null,"last_updated_by": null,"id": "eca07122-f4a2-42e7-b436-e7abb134b1d3"},{"login_id": "charlie3","user_name": "Charlie","role": null,"email": "charlie@example.com","person_id": "P003","is_superuser": true,"is_staff": true,"effective_date": "2023-03-01","expiry_date": "2025-12-31","deletion_mark": false,"deletion_reason": null,"deletion_date": null,"deletion_by": null,"create_date": "2025-09-15T16:56:58","creator_by": "231b2c7d-a2eb-47db-a40d-d400f7d250f6","last_updated_date": null,"last_updated_by": null,"id": "3822eb35-ecc7-494a-b992-344523c1ead2"}]},"faulty": {"count": 0,"details": []}}
}2.Accept-Language为zh
调整Header中Language值为zh
运行结果："message": "资料已存在：跳过3笔资料"
{"status_code": 200,"status": "success","message": "资料已存在：跳过3笔资料","record_count": 0,"data": {"created": {"count": 0,"details": []},"duplicates": {"count": 3,"details": [{"login_id": "alice1736","user_name": "Alice","role": null,"email": "alice@example.com","person_id": "P001","is_superuser": false,"is_staff": true,"effective_date": "2023-01-01","expiry_date": "2025-12-31","deletion_mark": false,"deletion_reason": null,"deletion_date": null,"deletion_by": null,"create_date": "2025-09-15T16:56:58","creator_by": "231b2c7d-a2eb-47db-a40d-d400f7d250f6","last_updated_date": null,"last_updated_by": null,"id": "bb6b9a96-239b-4956-a2a3-c39d9f6113fe"},{"login_id": "bob4","user_name": "Bob","role": null,"email": "bob@example.com","person_id": "P002","is_superuser": false,"is_staff": false,"effective_date": "2023-02-01","expiry_date": "2025-12-31","deletion_mark": true,"deletion_reason": null,"deletion_date": "2025-09-15T16:59:29","deletion_by": "231b2c7d-a2eb-47db-a40d-d400f7d250f6","create_date": "2025-09-15T16:59:29","creator_by": "231b2c7d-a2eb-47db-a40d-d400f7d250f6","last_updated_date": null,"last_updated_by": null,"id": "eca07122-f4a2-42e7-b436-e7abb134b1d3"},{"login_id": "charlie3","user_name": "Charlie","role": null,"email": "charlie@example.com","person_id": "P003","is_superuser": true,"is_staff": true,"effective_date": "2023-03-01","expiry_date": "2025-12-31","deletion_mark": false,"deletion_reason": null,"deletion_date": null,"deletion_by": null,"create_date": "2025-09-15T16:56:58","creator_by": "231b2c7d-a2eb-47db-a40d-d400f7d250f6","last_updated_date": null,"last_updated_by": null,"id": "3822eb35-ecc7-494a-b992-344523c1ead2"}]},"faulty": {"count": 0,"details": []}}
}3.Accept-Language为zh-TW
调整Header中Language值为zh-TW
运行结果："message": "資料已存在：跳過3筆資料"
{"status_code": 200,"status": "success","message": "資料已存在：跳過3筆資料","record_count": 0,"data": {"created": {"count": 0,"details": []},"duplicates": {"count": 3,"details": [{"login_id": "alice1736","user_name": "Alice","role": null,"email": "alice@example.com","person_id": "P001","is_superuser": false,"is_staff": true,"effective_date": "2023-01-01","expiry_date": "2025-12-31","deletion_mark": false,"deletion_reason": null,"deletion_date": null,"deletion_by": null,"create_date": "2025-09-15T16:56:58","creator_by": "231b2c7d-a2eb-47db-a40d-d400f7d250f6","last_updated_date": null,"last_updated_by": null,"id": "bb6b9a96-239b-4956-a2a3-c39d9f6113fe"},{"login_id": "bob4","user_name": "Bob","role": null,"email": "bob@example.com","person_id": "P002","is_superuser": false,"is_staff": false,"effective_date": "2023-02-01","expiry_date": "2025-12-31","deletion_mark": true,"deletion_reason": null,"deletion_date": "2025-09-15T16:59:29","deletion_by": "231b2c7d-a2eb-47db-a40d-d400f7d250f6","create_date": "2025-09-15T16:59:29","creator_by": "231b2c7d-a2eb-47db-a40d-d400f7d250f6","last_updated_date": null,"last_updated_by": null,"id": "eca07122-f4a2-42e7-b436-e7abb134b1d3"},{"login_id": "charlie3","user_name": "Charlie","role": null,"email": "charlie@example.com","person_id": "P003","is_superuser": true,"is_staff": true,"effective_date": "2023-03-01","expiry_date": "2025-12-31","deletion_mark": false,"deletion_reason": null,"deletion_date": null,"deletion_by": null,"create_date": "2025-09-15T16:56:58","creator_by": "231b2c7d-a2eb-47db-a40d-d400f7d250f6","last_updated_date": null,"last_updated_by": null,"id": "3822eb35-ecc7-494a-b992-344523c1ead2"}]},"faulty": {"count": 0,"details": []}}
}4.默认值zh
删除Header中Language，因为系统默认值为zh
运行结果："message": "资料已存在：跳过3笔资料"
5.默认值zh-TW
修改默认值zh-TW，终止项目重启。
运行结果："message": "資料已存在：跳過3筆資料"