《FastAPI零基础入门与进阶实战》第18篇:Token验证改善--CRUD中应用
系列文章目录
《FASTAPI零基础入门与进阶实战》
https://blog.csdn.net/sen_shan/category_12950843.html
第17篇:Token验证改善--与项目与用户关联https://blog.csdn.net/sen_shan/article/details/151707238
文章目录
目录
系列文章目录
文章目录
前言
login_manager
orm_curd
CRUD
router
测试
前言
上一章聚焦 Token 验证及项目-用户绑定，本章则围绕应用 API ID 与 User ID 的增、查、改、删展开。
login_manager
1.把login_manager.py从Models移转到schemas中，并且修改相关程序，此处不做说明；
2.修改\src\schemas\login_manager.py，新增class AuthManager
from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetimeclass AuthManager(BaseModel):api_id: struser_id: strusername: strrole: Optional[str] = Noneemail: strexp: datetimeapi_key: strorm_curd
修改src\core\orm_curd.py，本次修改内容比较多，附上全部代码：
# src\core\orm_curd.py
from typing import Set, Any, Dict, List, Optional, Type, TypeVar, Tuple, Union
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_idcreated, 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 = f"成功创建 {len(created_dicts)} 个{Model_Name}"if duplicates_dicts:message += f"，跳过 {len(duplicates_dicts)} 个重复{Model_Name}"if faulty:message += f"，有 {len(faulty)} 个数据格式错误"response_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_id# 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}return retMes.Success(payload,f'共计：{len(details)} 条数据',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_id# 1. 若未显式传入 pk_values，则从 schema_obj 提取if strU.is_empty(pk_values):if schema_obj is None:retMes.Raise(message="schema_obj 为空，无法提取主键值").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):retMes.Raise(message="主键值列表为空或包含 None").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.if update_obj is None:return retMes.Error("修改内容Json 为空").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)return retMes.Success(updated_entities,f'共计：{len(updated_entities)} 条数据',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_id# 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):retMes.Raise(message="主键值列表为空或包含 None").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"""# 1.if delete_obj is None:return retMes.Error("删除内容为空").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:retMes.Raise(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_parts.append(f"成功删除 {len(deleted_entities)} 笔资料")if marked_entities:message_parts.append(f"成功标注失效记录 {len(marked_entities)}")message = "，".join(message_parts) or "无任何变动"response_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'{message} ',record_count).mes()
所有代码参数中，增加auth: AuthManager，后期获取api_id与User_id.
新增
在写入数据库前，统一为实体补全审计字段与逻辑删除标记，保证数据可追溯、可复用。
❶. api_id与User_id获取
根据登录信息获取 api_id与User_id。
❷.项目隔离
若模型带 api_id 列，自动把当前登录用户的项目编号写入，防止跨项目数据串扰。
❸. 创建人 / 更新人
新增时： creator_by = 当前用户 ID；
首次写入时： last_updated_by 、 last_updated_date 置空，由数据库触发器或后续更新逻辑回填。
❹. 逻辑删除
默认 deletion_mark = 0 ， deletion_date 与 deletion_by 置空；
当调用方显式把 deletion_mark 设为 1 时，自动写入删除时间（当前时间）及删除人，实现“谁删的、何时删”一目了然。
❺. 防御式编程
全程用 hasattr 判断字段是否存在，同一段代码可安全应用于不同表，无字段则跳过，避免 AttributeError 。
修改
更新前自动补全「项目隔离」与「审计字段」，同一段代码可安全复用到任意表。
❶. 项目隔离
若模型存在 api_id 列，则自动带上 api_id = 当前登录项目 ID 的条件，防止跨项目误改数据；全局表无此字段则跳过，直接查全量。
❷. 审计字段
只要表里有 last_updated_date ，就写入当前时间；
有 last_updated_by ，就写入当前用户 ID。
不存在对应字段的表自动忽略，避免抛错。
❸. 防御式写法
全程用 hasattr 判断，模型增减字段无需改动公共逻辑，一次编写，全库通用。
删除
删除除了参加参数，程序逻辑变化不大。
查询
查询前自动按「api_id 」隔离数据，同一段代码兼容所有模型。
❶. 若模型存在 api_id 字段，则自动追加过滤条件 api_id = 当前登录项目 ID ，确保仅返回本项目数据。
❷. 若模型无 api_id 字段（如系统级字典表），则跳过过滤，直接查全量，避免报错。
❸. 全程使用 hasattr 判断，模型增减字段无需调整公共逻辑，实现“一次编写，全库通用」。
CRUD
修改crud/sys_user.py，本改版比较多，附上全部代码：
# crud/sys_user.py
from pydoc import pagerfrom sqlalchemy.orm import Session
from src.models import sys_user as models
from src.schemas import sys_user as schemas, request_model
from src.core import retMes, str_utils as strU, orm_curd
from src.schemas.login_manager import AuthManagerdef create_user(db: Session,auth: AuthManager,user: schemas.SysUserCreate):"""新增单个用户。参数说明----------db : Session数据库会话对象。auth: 登录信息user : schemas.SysUserCreate待创建用户的 Pydantic 模型实例。返回值-------dict按 retMes.Success 包装好的 dict备注----内部调用 create_users 实现单条批量插入，行为与批量接口保持一致。"""# 检查用户是否已存在（基于login_id）return orm_curd.Insert(db,auth,models.SysUser,schemas.SysUser,"用户",[user.dict()], # [u.dict() for u in users], # users,["api_id", "login_id"],skip_duplicates=True)def create_users(db: Session,auth: AuthManager,users: list[schemas.SysUserCreate]):"""批量新增用户。参数说明----------db : Session数据库会话对象。auth: 登录信息users : list[schemas.SysUserCreate]待创建用户的 Pydantic 模型实例列表。返回值-------dict按 retMes.Success 包装好的 dict"""return orm_curd.Insert(db,auth,models.SysUser,schemas.SysUser,"用户",users, #[u.dict() for u in users], # users,["api_id", "login_id"],skip_duplicates=True)def get_users(db: Session,auth: AuthManager,page: int = 0,page_size: int = 100,FilterModel: request_model.QueryModel = None,id: str = None):"""分页查询系统用户列表。参数说明----------db : Session数据库会话对象，用于执行 ORM 操作。auth: 登录信息authManager:authManager登录信息page : int, 可选页码，从 0 开始计数；默认值为 0。page_size : int, 可选每页返回的记录数；默认值为 100。FilterModel : request_model.QueryModel, 可选查询条件模型，用于构造过滤条件；默认值为 None，表示不过滤。id : str, 可选用户 ID 精准匹配；默认值为 None，表示不根据 ID 过滤。返回值-------dict按 retMes.Success 包装好的 dict备注说明--------1. 本函数通过 `orm_curd.query_with_page` 统一封装，自动完成分页、过滤、排序等操作。2. 当 `id` 参数非空时，会优先使用精准匹配；其余过滤条件以 `FilterModel` 为准。3. 若 `schemas.SysUser` 为 None，则返回原始 ORM 对象，跳过 Pydantic 校验，性能略高。4. 调用方需确保传入的 `db` 处于活跃事务中，避免 Lazy Load 异常。"""return orm_curd.query_with_page(db,auth,models.SysUser,schemas.SysUser, # 如不需要字段校验可传 Noneid=id,filter_model=FilterModel,page=page,page_size=page_size)def update_user(db: Session,auth: AuthManager,user_id: str,user: schemas.SysUser,) -> schemas.SysUser:"""更新单个用户参数----db : SQLAlchemy Sessionauth: 登录信息user_id : 待更新用户的主键 iduser : Pydantic 模型，仅包含需要修改的字段（exclude_unset=True）返回----更新后的用户 ORM 对象（已 flush，含最新值）"""update_model = request_model.UpdateModel([request_model.UpdateItem(pk_names=["id"],pk_values=[user_id],data=user.dict(exclude_unset=True))])return orm_curd.update(db,auth,models.SysUser,update_model)def update_users(db: Session,auth: AuthManager,updateModel: request_model.UpdateModel):"""批量更新用户参数----db : SQLAlchemy Sessionauth: 登录信息updateModel : request_model.UpdateModel待更新的用户列表，包含主键 id 和数据字段返回----按 retMes.Success 封装的 dict"""return orm_curd.update(db,auth,models.SysUser,updateModel)def delete_user(db: Session,auth: AuthManager,deleteModel: request_model.DeleteModel):"""调用通用 delete 接口删除指定用户（硬删除）:param db: SQLAlchemy 会话:param auth: 登录信息:param deleteModel: 删除内容:return: 统一响应格式"""# 直接调用通用删除函数return orm_curd.delete(db=db,auth=auth,model_cls=models.SysUser,delete_obj=deleteModel)
在所有程序增加参数auth: AuthManager登录信息
router
修改src/router/sys_user.py，，本改版比较多，附上全部代码：
# src/router/sys_user.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)
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)
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, retMesrouter = APIRouter( # prefix="/user",# tags=["sys_user"] # ,# dependencies=[Depends(get_token_header)],# responses={404: {"description": "Not found"}},# description="Operations related to system users",# on_startup=[startup_event],# on_shutdown=[shutdown_event]
)# 创建用户
#@router.post("/users/", response_model=list[sysUserSchema.SysUser])
@router.post("/users/", response_model=request_model.ResponseModel)
def create_users(users: list[sysUserSchema.SysUserCreate],db: Session = Depends(get_db),auth: AuthManager = Depends(dependencies.auth_token)):return sysUserCrud.create_users(db,auth, users)# @router.post("/user/", response_model=sysUserSchema.SysUser)
@router.post("/user/", response_model=request_model.ResponseModel)
def create_user(user: sysUserSchema.SysUserCreate,db: Session = Depends(get_db),auth: AuthManager = Depends(dependencies.auth_token)):return sysUserCrud.create_user(db,auth, user)# 获取用户
@router.get("/users/{id}", response_model=request_model.ResponseModel)
def read_user(id: str,db: Session = Depends(get_db),auth: AuthManager = Depends(dependencies.auth_token)
):# 按 id 查单条db_user = sysUserCrud.get_users(db, auth, None, None, None, id)if not db_user: # 查不到retMes.Raise(message="User not found").mes() # 统一 401return db_user# 获取所有用户
# @router.get("/users/", response_model=list[sysUserSchema.SysUser])
@router.get("/users/", response_model=request_model.ResponseModel)
def read_users(page: int = 1,page_size: int = 100,db: Session = Depends(get_db),auth: AuthManager = Depends(dependencies.auth_token)):# 将页码转换为skip值# skip = (page - 1) * page_sizereturn sysUserCrud.get_users(db, auth, page, page_size)# 获取所有用户
# @router.post("/users/", response_model=list[sysUserSchema.SysUser])
@router.post("/users/query/", response_model=request_model.ResponseModel)
def read_users(queryModel: request_model.QueryModel = None,db: Session = Depends(get_db),auth: AuthManager = Depends(dependencies.auth_token)):# 将页码转换为skip值return sysUserCrud.get_users(db, auth, None, None, queryModel)# 更新用户
@router.put("/users/{user_id}", response_model=request_model.ResponseModel)
def update_user(user_id: str,user: sysUserSchema.SysUser,db: Session = Depends(get_db),auth: AuthManager = Depends(dependencies.auth_token)):db_user = sysUserCrud.update_user(db, auth, user_id, user)if db_user is None:retMes.Raise(message="User not found").mes() # 统一 401return db_user@router.put("/users/", response_model=request_model.ResponseModel)
def update_user(updateModel: request_model.UpdateModel,db: Session = Depends(get_db),auth: AuthManager = Depends(dependencies.auth_token)):db_user = sysUserCrud.update_users(db, auth, updateModel)if db_user is None:retMes.Raise(message="User not found").mes() # 统一 401print(db_user)return db_user# 删除用户
@router.delete("/users/", response_model=request_model.ResponseModel)
def delete_user(deleteModel: request_model.DeleteModel,db: Session = Depends(get_db),auth: AuthManager = Depends(dependencies.auth_token)):db_user = sysUserCrud.delete_user(db,auth, deleteModel)if db_user is None:retMes.Raise(message="User not found").mes() # 统一 401return db_user在所有程序增加参数auth: AuthManager = Depends(dependencies.auth_token)获取登录信息
测试
在Header 增加x-api-key与token，并属于正确值，具体参考第七章节中验证Token。