refactor: 重构系统配置和用户模型相关代码

feat: 添加中间件系统配置获取功能
fix: 修复用户模型和CRUD基础类的问题
docs: 更新README和界面文本
style: 清理无用代码和注释
This commit is contained in:
zhangtao
2025-10-08 01:32:14 +08:00
parent 319a332ad3
commit e669722706
17 changed files with 232 additions and 189 deletions
@@ -12,9 +12,9 @@ from app.core.redis_crud import RedisCURD
from app.core.exceptions import CustomException
from app.core.logger import logger
from app.api.v1.module_system.auth.schema import AuthSchema
from app.api.v1.module_system.dict.schema import DictDataCreateSchema,DictDataOutSchema,DictDataUpdateSchema,DictTypeCreateSchema,DictTypeOutSchema,DictTypeUpdateSchema
from app.api.v1.module_system.dict.param import DictDataQueryParam, DictTypeQueryParam
from app.api.v1.module_system.dict.crud import DictDataCRUD, DictTypeCRUD
from .schema import DictDataCreateSchema,DictDataOutSchema,DictDataUpdateSchema,DictTypeCreateSchema,DictTypeOutSchema,DictTypeUpdateSchema
from .param import DictDataQueryParam, DictTypeQueryParam
from .crud import DictDataCRUD, DictTypeCRUD
class DictTypeService:
@@ -216,4 +216,70 @@ class ParamsService:
logger.error(f"解析系统配置数据失败: {e}")
continue
return configs
return configs
@classmethod
async def get_system_config_for_middleware(cls, redis: Redis) -> Dict[str, Any]:
"""获取中间件所需的系统配置
返回:
Dict: 包含演示模式、IP白名单、API白名单和IP黑名单的配置字典
"""
# 定义需要获取的配置键
config_keys = [
f"{RedisInitKeyConfig.SYSTEM_CONFIG.key}:demo_enable",
f"{RedisInitKeyConfig.SYSTEM_CONFIG.key}:ip_white_list",
f"{RedisInitKeyConfig.SYSTEM_CONFIG.key}:white_api_list_path",
f"{RedisInitKeyConfig.SYSTEM_CONFIG.key}:ip_black_list"
]
# 批量获取配置
config_values = await RedisCURD(redis).mget(config_keys)
# 初始化默认配置
config_result = {
"is_demo_mode": False,
"demo_ip_white_list": [],
"api_white_list": [],
"ip_black_list": []
}
# 解析演示模式配置
if config_values[0]:
try:
demo_config = json.loads(config_values[0])
config_result["is_demo_mode"] = demo_config.get("config_value", False) if isinstance(demo_config, dict) else False
except json.JSONDecodeError:
logger.error(f"解析演示模式配置失败")
# 解析IP白名单配置
if config_values[1]:
try:
ip_white_config = json.loads(config_values[1])
demo_ip_white_list = ip_white_config.get("config_value", []) if isinstance(ip_white_config, dict) else []
# 确保是列表类型
config_result["demo_ip_white_list"] = demo_ip_white_list if isinstance(demo_ip_white_list, list) else []
except json.JSONDecodeError:
logger.error(f"解析IP白名单配置失败")
# 解析API路径白名单
if config_values[2]:
try:
white_api_config = json.loads(config_values[2])
api_white_list = white_api_config.get("config_value", []) if isinstance(white_api_config, dict) else []
# 确保是列表类型
config_result["api_white_list"] = api_white_list if isinstance(api_white_list, list) else []
except json.JSONDecodeError:
logger.error(f"解析API白名单配置失败")
# 解析IP黑名单
if config_values[3]:
try:
black_ip_config = json.loads(config_values[3])
ip_black_list = black_ip_config.get("config_value", []) if isinstance(black_ip_config, dict) else []
# 确保是列表类型
config_result["ip_black_list"] = ip_black_list if isinstance(ip_black_list, list) else []
except json.JSONDecodeError:
logger.error(f"解析IP黑名单配置失败")
return config_result
+11 -2
View File
@@ -7,7 +7,7 @@
from datetime import datetime
from typing import Optional, List
from sqlalchemy import Boolean, String, Integer, DateTime, ForeignKey
from sqlalchemy import Boolean, String, Integer, DateTime, ForeignKey, Text
from sqlalchemy.orm import relationship, Mapped, mapped_column
from app.api.v1.module_system.dept.model import DeptModel
@@ -62,13 +62,15 @@ class UserPositionsModel(MappedBase):
)
class UserModel(CreatorMixin):
class UserModel(MappedBase):
"""
用户模型
"""
__tablename__ = "system_users"
__table_args__ = ({'comment': '用户表'})
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True, comment='主键ID')
username: Mapped[str] = mapped_column(String(32),nullable=False,unique=True,comment="用户名/登录账号")
password: Mapped[str] = mapped_column(String(255),nullable=False,comment="密码哈希")
name: Mapped[str] = mapped_column(String(32),nullable=False,comment="昵称")
@@ -84,3 +86,10 @@ class UserModel(CreatorMixin):
dept: Mapped[Optional["DeptModel"]] = relationship(back_populates="users",foreign_keys=[dept_id],lazy="selectin")
roles: Mapped[List["RoleModel"]] = relationship(secondary="system_user_roles",back_populates="users",lazy="selectin")
positions: Mapped[List["PositionModel"]] = relationship(secondary="system_user_positions",back_populates="users",lazy="selectin")
description: Mapped[Optional[str]] = mapped_column(Text, nullable=True, default=None, comment="备注/描述")
created_at: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True, default=datetime.now, comment='创建时间')
updated_at: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True, default=datetime.now, onupdate=datetime.now, comment='更新时间')
creator_id: Mapped[Optional[int]] = mapped_column(Integer, ForeignKey('system_users.id', ondelete="SET NULL", onupdate="CASCADE"), nullable=True, index=True, comment="创建人ID")
creator: Mapped[Optional["UserModel"]] = relationship(foreign_keys=[creator_id],lazy="selectin",remote_side=[id])
@@ -83,9 +83,8 @@ class UserService:
# 创建用户
if data.password:
data.password = PwdUtil.set_password_hash(password=data.password)
# user_dict = data.model_dump(exclude_unset=True, exclude={"role_ids", "position_ids"})
# new_user = await UserCRUD(auth).create(data=user_dict)
new_user = await UserCRUD(auth).create(data=data)
user_dict = data.model_dump(exclude_unset=True, exclude={"role_ids", "position_ids"})
new_user = await UserCRUD(auth).create(data=user_dict)
# 设置角色和岗位
if data.role_ids and len(data.role_ids) > 0:
@@ -175,7 +174,7 @@ class UserService:
async def get_current_user_info_service(cls, auth: AuthSchema) -> Dict:
"""获取当前用户信息"""
# 获取用户基本信息
if not auth.user:
if not auth.user or not auth.user.id:
raise CustomException(msg="用户不存在")
user = await UserCRUD(auth).get_by_id_crud(id=auth.user.id)
# 获取部门名称
@@ -210,7 +209,7 @@ class UserService:
@classmethod
async def update_current_user_info_service(cls, auth: AuthSchema, data: CurrentUserUpdateSchema) -> Dict:
"""更新当前用户信息"""
if not auth.user:
if not auth.user or not auth.user.id:
raise CustomException(msg="用户不存在")
user = await UserCRUD(auth).get_by_id_crud(id=auth.user.id)
if not user:
@@ -247,7 +246,7 @@ class UserService:
@classmethod
async def change_user_password_service(cls, auth: AuthSchema, data: UserChangePasswordSchema) -> Dict:
"""修改用户密码"""
if not auth.user:
if not auth.user or not auth.user.id:
raise CustomException(msg="用户不存在")
if not data.old_password or not data.new_password:
raise CustomException(msg='密码不能为空')