当前位置：首页 > news >正文

《FastAPI零基础入门与进阶实战》第19篇:消息管理

news 2025/9/19 8:03:13

系列文章目录

《FASTAPI零基础入门与进阶实战》https://blog.csdn.net/sen_shan/category_12950843.html第18篇:Token验证改善--CRUD中应用https://blog.csdn.net/sen_shan/article/details/151722431?spm=1011.2415.3001.5331

文章目录

系列文章目录

文章目录

前言

Models

schemas

CURD

Router

运行

前言

对消息统一封装消息内容，把code→msg替换，同时根据请求头Accept-Language返回中/英/繁体等多语满足国际化、运维、合规三方面的统一消息管理需求。

本章节主要是Models,Schemas,CRUD,Routerg构建，下一章节对初始化与消息封装。

Models

新建src\models\sys_message.py文件。

# src\models\sys_message.py
from sqlalchemy import Column, String, Date, DateTime, Boolean, Integer,UniqueConstraint
from src.core.ormdb import Base
from src.core import str_utils as strUclass SysMessage(Base):__tablename__ = "sys_message"__table_args__ = (UniqueConstraint("api_id", "language", "code",name="uk_sys_message_api_code_lang"),{"comment": "多国语言消息表"},)id = Column(String(36), primary_key=True, default=lambda: strU.generate_id(), comment="唯一标识ID")api_id = Column(String(36), nullable=True, comment="API主键，用于关联其他系统")language = Column(String(10), nullable=True, comment="语言标识 zh/en/fr...")code = Column(String(40), nullable=True, comment="错误码")description = Column(String(255), nullable=True, comment='描述')message = Column(String(500), nullable=True, comment="翻译文本")remark = Column(String(255), nullable=True,comment='业务备注，记录该条翻译的使用场景或变更原因')deletion_mark = Column(Boolean, default=False, comment="删除标志，0表示未删除，1表示已删除")deletion_reason = Column(String(255), nullable=True, comment="删除原因，记录删除的具体原因")deletion_date = Column(DateTime, nullable=True, comment="删除日期，记录删除的时间")deletion_by = Column(String(255), nullable=True, comment="删除者，记录删除操作的执行者")# create_date = Column(DateTime, default="now()", comment="创建日期，记录创建的时间")create_date = Column(DateTime, default=lambda: strU.now(), comment="创建日期，记录创建的时间")creator_by = Column(String(255), nullable=True, comment="创建者，记录创建操作的执行者")last_updated_date = Column(DateTime, nullable=True, comment="最后更新时间，记录最后更新的时间")last_updated_by = Column(String(255), nullable=True, comment="最后更新者，记录最后更新操作的执行者")

功能说明：

SysMessage 表集中存放系统内所有 API 的多国语言提示文本。

以 api_id+language+code 做唯一键，保证同一错误码在同一语言下仅存在一条有效翻译。

通过 code 与调用方解耦：服务层只返回错误码，AOP 拦截后根据请求头 Accept-Language 动态查询本表 message 字段，实时替换提示语，支持 zh/en/fr/zh-TW 等任意语种扩展。

description 记录含义

remark 记录业务场景

deletion** 字段实现逻辑删除与审计

create** / last_updated** 提供全程追踪，满足国际化、运维、合规三方面的统一消息管理需求。

schemas

新建src/schemas/sys_message.py文件

# src/schemas/sys_message.py
from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetimeclass SysMessageBase(BaseModel):language: str = Field(..., max_length=10, description="语言标识 zh/en/fr...")code: str = Field(..., description="错误码")description: Optional[str] = Field(..., max_length=255, description="描述")message: str = Field(..., max_length=500, description="翻译文本")remark: Optional[str] = Field(None, max_length=500,description='业务备注')class SysMessageCreate(SysMessageBase):api_id: Optional[str] = Field(None, description="API主键，用于关联其他系统")class SysMessage(SysMessageBase):id: strapi_id: Optional[str]deletion_mark: bool = Falsedeletion_reason: Optional[str]deletion_date: Optional[datetime]deletion_by: Optional[str]create_date: datetimecreator_by: Optional[str]last_updated_date: Optional[datetime]last_updated_by: Optional[str]class Config:from_attributes = True      # Pydantic V2 新写法

功能说明
sys_message.py 定义了 SysMessage 表的三层 Pydantic 模型，用于“国际化消息”全生命周期管理。
1.  SysMessageBase 抽象最小字段集，确保新增/编辑时必须提供语言、错误码、翻译与描述，前端表单与后端参数同时复用，避免字段漂移。
2.  SysMessageCreate 在 Base 基础上可选携带 api_id ，满足“先录消息、后绑定接口”的敏捷流程；序列化时自动校验长度与空值，直接由 CRUD 接口接收。
3.  SysMessage 继承 Base 并追加主键、审计、删除标记等只读字段，配合 from_attributes=True 可一键把 SQLAlchemy 实体转为响应 DTO，供列表、单条、删除记录接口统一返回，实现“入库模型—业务模型—出参模型”三层一致，降低重复代码与维护成本。

CURD

# src/crud/sys_message.py
from typing import List, Optional
from sqlalchemy.orm import Session
from src.models import sys_message as models
from src.schemas import sys_message as schemas, request_model
from src.core import retMes, str_utils as strU, orm_curd
from src.schemas.login_manager import AuthManager
from src.crud import sys_api_project as SysApiProjectCRUD
from src.core.config import Config
import json, os# ---------------------- 单条新增 ----------------------
def create_message(db: Session,auth: AuthManager,msg: schemas.SysMessageCreate) -> dict:"""新增单条语言消息。内部调用 create_messages 保持行为一致。"""return orm_curd.Insert(db,auth,models.SysMessage,schemas.SysMessageInDB,"语言消息",[msg.dict()],["api_id", "language", "code"],  # 业务唯一键skip_duplicates=True)# ---------------------- 批量新增 ----------------------
def create_messages(db: Session,auth: AuthManager,messages: List[schemas.SysMessageCreate]) -> dict:"""批量新增语言消息。"""return orm_curd.Insert(db,auth,models.SysMessage,schemas.SysMessageInDB,"语言消息",messages,["api_id", "language", "code"],skip_duplicates=True)# ---------------------- 分页查询 ----------------------
def get_messages(db: Session,auth: AuthManager,page: int = 0,page_size: int = 100,FilterModel: Optional[request_model.QueryModel] = None,id: Optional[str] = None) -> dict:"""分页查询语言消息列表。"""return orm_curd.query_with_page(db,auth,models.SysMessage,schemas.SysMessageInDB,  # 如需原始对象可传 Noneid=id,filter_model=FilterModel,page=page,page_size=page_size)# ---------------------- 单条更新 ----------------------
def update_message(db: Session,auth: AuthManager,msg_id: str,msg: schemas.SysMessage) -> schemas.SysMessage:"""更新单条语言消息。"""update_model = request_model.UpdateModel([request_model.UpdateItem(pk_names=["id"],pk_values=[msg_id],data=msg.dict(exclude_unset=True))])return orm_curd.update(db,auth,models.SysMessage,update_model)# ---------------------- 批量更新 ----------------------
def update_messages(db: Session,auth: AuthManager,updateModel: request_model.UpdateModel) -> dict:"""批量更新语言消息。"""return orm_curd.update(db,auth,models.SysMessage,updateModel)# ---------------------- 删除 ----------------------
def delete_messages(db: Session,auth: AuthManager,deleteModel: request_model.DeleteModel) -> dict:"""删除语言消息（硬删除）。"""return orm_curd.delete(db=db,auth=auth,model_cls=models.SysMessage,delete_obj=deleteModel)

功能性说明
sys_message.py 封装了“国际化消息”完整的 CRUD 行为，向上承接 REST/GraphQL 接口，向下屏蔽 ORM 细节，提供以下能力：
1. 单条与批量新增
统一调用 orm_curd.Insert ，以 (api_id, language, code) 为业务唯一键，自动跳过重复，保证同一错误码在同一语言下只有一条有效文案。
2. 分页查询
支持通用过滤模型 QueryModel ，可按语言、错误码、api_id 等字段自由组合，返回分页数据与总条数，满足后台翻译管理界面快速检索需求。
3. 单条与批量更新
使用 UpdateModel 结构，支持“主键+变更字段”模式，只更新传入的非空字段，自动写入 last_updated_by/date ，兼顾灵活与审计。
4. 物理删除
提供硬删除接口，传入 DeleteModel 即可批量删除；若后续需要软删除，仅需替换 orm_curd.delete 实现，无需改动上层调用。
5. 统一审计
所有写操作强制传入 AuthManager ，自动把当前登录人写入 creator_by / last_updated_by / deletion_by ，实现“谁新增、谁修改、谁删除”全程可追溯。
通过复用底层 orm_curd 模板，本文件将重复 SQL 收敛到一处，使国际化消息维护工作聚焦于业务配置而非数据访问，大幅提升可维护性与发布效率。

Router

修改src/router/sys_manges.py

增加sys_message 引用

# src/router/sys_manges.py
# from main import startup_event, shutdown_event
from src.crud import (sys_user as sysUserCrud,sys_api_project as SysApiProjectCrud,sys_api_teamwork as SysApiTeamworkCrud,sys_field_mapping as SysFieldMappingCrud,sys_organization as SysOrganizationCrud,sys_table_mapping as SysTableMappingCrud,sys_message as sysMessageCrud)
from src.schemas import (sys_user as sysUserSchema,sys_api_project as SysApiProjectSchema,sys_api_teamwork as SysApiTeamworkSchema,sys_field_mapping as SysFieldMappingSchema,sys_organization as SysOrganizationSchema,sys_table_mapping as SysTableMappingSchema,sys_message as sysMessageSchema)
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from src.core.ormdb import get_db
from src.schemas import request_model
from src.schemas.login_manager import AuthManager
from src.core import dependencies, retMes

增加消息管理Router


# ========== 语言消息管理 ==========
@router.post("/messages/", response_model=request_model.ResponseModel)
def create_messages(messages: list[sysMessageSchema.SysMessageCreate],db: Session = Depends(get_db),auth: AuthManager = Depends(dependencies.auth_token)
):return sysMessageCrud.create_messages(db, auth, messages)@router.post("/message/", response_model=request_model.ResponseModel)
def create_message(message: sysMessageSchema.SysMessageCreate,db: Session = Depends(get_db),auth: AuthManager = Depends(dependencies.auth_token)
):return sysMessageCrud.create_message(db, auth, message)@router.get("/messages/", response_model=request_model.ResponseModel)
def read_messages(page: int = 1,page_size: int = 100,db: Session = Depends(get_db),auth: AuthManager = Depends(dependencies.auth_token)
):return sysMessageCrud.get_messages(db, auth, page, page_size)@router.post("/messages/query/", response_model=request_model.ResponseModel)
def read_messages_query(queryModel: request_model.QueryModel = None,db: Session = Depends(get_db),auth: AuthManager = Depends(dependencies.auth_token)
):return sysMessageCrud.get_messages(db, auth, filterModel=queryModel)@router.put("/messages/{msg_id}", response_model=request_model.ResponseModel)
def update_message(msg_id: str,message: sysMessageSchema.SysMessage,db: Session = Depends(get_db),auth: AuthManager = Depends(dependencies.auth_token)
):return sysMessageCrud.update_message(db, auth, msg_id, message)@router.put("/messages/", response_model=request_model.ResponseModel)
def update_messages(updateModel: request_model.UpdateModel,db: Session = Depends(get_db),auth: AuthManager = Depends(dependencies.auth_token)
):return sysMessageCrud.update_messages(db, auth, updateModel)@router.delete("/messages/", response_model=request_model.ResponseModel)
def delete_messages(deleteModel: request_model.DeleteModel,db: Session = Depends(get_db),auth: AuthManager = Depends(dependencies.auth_token)
):return sysMessageCrud.delete_messages(db, auth, deleteModel)

功能性说明
本组接口以 /messages 为核心，提供“国际化语言消息”全生命周期的 REST 管理能力，具体功能如下：
1. 新增
POST /message/ 单条录入，方便后台快速添加；
POST /messages/ 批量录入，支持一次导入多语言、多错误码，自动去重（api_id+language+code），用于初始化或 Excel 导入。
2. 查询
GET /messages/ 分页取全表，页码与条数可传，适用于翻译管理首页；
POST /messages/query/ 按条件过滤，请求体使用 QueryModel ，可按语言、错误码、api_id、描述等任意字段组合查询，满足运维快速定位文案。
3. 更新
PUT /messages/{msg_id} 单条全量更新，用于编辑单行翻译；
PUT /messages/ 批量部分字段更新，使用 UpdateModel 结构，只传需要改动的列，系统自动写入最新操作人与时间，支持批量上线、批量改备注等场景。
4. 删除
DELETE /messages/ 物理删除，传入 DeleteModel 可一次删除多条，后台自动记录操作者，满足“错误码废弃”或“语言下线”需求。
所有接口统一返回 ResponseModel ，自带 code/msg/data 封装；依赖注入 AuthManager 实现登录态校验与操作审计，无需每个方法重复鉴权。通过本组接口，前端翻译管理界面可在零 SQL 的情况下完成“增删改查+导入导出”，实现系统提示语的集中、可控、可追溯的国际化运营。