refactor: P3c 服务层迁入 src/modules/ → service.py（29 服务）
- 22 个单服务 + 3 组合并（config+config_item、dict+dict_item、login_log+operation_log） - 特殊命名：job_executor→executor、scheduler→scheduler_service、 mail_svc→mail_service、sms_svc→sms_service（含使用处标识符改写） - 重写引用：跨服务（module 别名/子模块）、models/forms→src.modules、 db/extensions→src.core、utils→src.common|core、R→src.core.response - application/services/ 全部替换为 shim 【验收】全部服务文件编译通过；应用启动；pytest 5 passed；核心服务导入正常 Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
DjangoAdmin Dev
co-authored by
Claude
Claude
parent
2586c0aace
commit
87aec4e26e
@@ -1,238 +1,8 @@
|
||||
"""
|
||||
用户角色关联服务模块
|
||||
兼容 shim（绞杀者迁移期间保留）
|
||||
=============================================================
|
||||
|
||||
该模块实现了用户与角色之间的关联管理功能。
|
||||
包括根据用户ID查询角色列表等操作。
|
||||
|
||||
命名规范：
|
||||
- 函数参数使用驼峰命名（如 userId）
|
||||
- 返回数据使用驼峰命名（如 roleList）- 与前端规范保持一致
|
||||
|
||||
功能模块：
|
||||
1. 角色列表查询：根据用户ID获取该用户已分配的角色列表
|
||||
服务已迁移至 src.modules.association.user_role.service。
|
||||
本文件仅作兼容转发，最终在 P7（切尾）阶段删除。
|
||||
"""
|
||||
|
||||
import logging
|
||||
|
||||
from application.models.role import Role
|
||||
from application.models.user import User
|
||||
from application.models.user_role import UserRole
|
||||
from extensions import db
|
||||
from utils.request import parse_id_list
|
||||
|
||||
# 获取日志记录器
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# ============================================================
|
||||
# 用户角色列表查询
|
||||
# ============================================================
|
||||
|
||||
def get_user_role_list(user_id):
|
||||
"""
|
||||
根据用户ID查询角色列表
|
||||
|
||||
通过关联表 user_role 查询指定用户已分配的角色信息。
|
||||
只返回未被删除的角色（is_delete == 0）。
|
||||
|
||||
Args:
|
||||
user_id: 用户ID
|
||||
|
||||
Returns:
|
||||
list: 角色列表，每个元素包含 id 和 name，无数据时返回空列表
|
||||
格式示例：[{"id": 1, "name": "管理员"}, {"id": 2, "name": "普通用户"}]
|
||||
|
||||
Example:
|
||||
role_list = get_user_role_list(100)
|
||||
for role in role_list:
|
||||
print(role['id'], role['name'])
|
||||
"""
|
||||
# 参数有效性验证
|
||||
if not user_id or user_id <= 0:
|
||||
return []
|
||||
|
||||
try:
|
||||
# ORM JOIN查询，表名自动从Model获取
|
||||
result = (
|
||||
db.session.query(Role.id, Role.name)
|
||||
.join(UserRole, Role.id == UserRole.role_id)
|
||||
.filter(UserRole.user_id == user_id, Role.is_delete == 0)
|
||||
.all()
|
||||
)
|
||||
|
||||
# 构建角色列表
|
||||
role_list = [{'id': row.id, 'name': row.name} for row in result]
|
||||
|
||||
return role_list
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"查询用户角色列表异常，用户ID: {user_id}, 错误: {str(e)}")
|
||||
return []
|
||||
|
||||
|
||||
# ============================================================
|
||||
# 用户角色关联添加/更新
|
||||
# ============================================================
|
||||
|
||||
def save_user_roles(user_id, role_ids):
|
||||
"""
|
||||
保存用户角色关联关系
|
||||
|
||||
先删除用户原有的所有角色关联，再重新创建新的关联关系。
|
||||
|
||||
Args:
|
||||
user_id: 用户ID
|
||||
role_ids: 角色ID列表，如 [1, 2, 3] 或逗号分隔的字符串 "1,2,3"
|
||||
|
||||
Returns:
|
||||
bool: 操作成功返回 True，失败返回 False
|
||||
|
||||
Example:
|
||||
# 使用列表
|
||||
save_user_roles(100, [1, 2, 3])
|
||||
# 使用逗号分隔字符串
|
||||
save_user_roles(100, "1,2,3")
|
||||
"""
|
||||
# 参数有效性验证
|
||||
if not user_id or user_id <= 0:
|
||||
logger.warning(f"保存用户角色关联失败：用户ID无效 {user_id}")
|
||||
return False
|
||||
|
||||
try:
|
||||
# 解析角色ID列表
|
||||
if isinstance(role_ids, str):
|
||||
if not role_ids:
|
||||
role_id_list = []
|
||||
else:
|
||||
role_id_list = parse_id_list(role_ids)
|
||||
elif isinstance(role_ids, list):
|
||||
role_id_list = role_ids
|
||||
else:
|
||||
role_id_list = []
|
||||
|
||||
# 删除用户原有角色关系
|
||||
UserRole.query.filter(UserRole.user_id == user_id).delete()
|
||||
|
||||
# 创建新的用户角色关系
|
||||
for role_id in role_id_list:
|
||||
if not role_id:
|
||||
continue
|
||||
user_role = UserRole(
|
||||
user_id=user_id,
|
||||
role_id=role_id
|
||||
)
|
||||
db.session.add(user_role)
|
||||
|
||||
db.session.flush()
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"保存用户角色关联异常，用户ID: {user_id}, 错误: {str(e)}")
|
||||
return False
|
||||
|
||||
|
||||
# ============================================================
|
||||
# 用户角色删除
|
||||
# ============================================================
|
||||
|
||||
def delete_user_roles_by_user_id(user_id):
|
||||
"""
|
||||
删除指定用户的所有角色关联
|
||||
|
||||
Args:
|
||||
user_id: 用户ID
|
||||
|
||||
Returns:
|
||||
bool: 操作成功返回 True，失败返回 False
|
||||
"""
|
||||
if not user_id or user_id <= 0:
|
||||
logger.warning(f"删除用户角色关联失败：用户ID无效 {user_id}")
|
||||
return False
|
||||
|
||||
try:
|
||||
count = UserRole.query.filter(UserRole.user_id == user_id).delete()
|
||||
db.session.flush()
|
||||
|
||||
logger.info(f"删除用户角色关联成功，用户ID: {user_id}, 删除数量: {count}")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"删除用户角色关联异常，用户ID: {user_id}, 错误: {str(e)}")
|
||||
return False
|
||||
|
||||
|
||||
# ============================================================
|
||||
# 用户角色批量查询
|
||||
# ============================================================
|
||||
|
||||
def get_user_roles_batch(user_ids):
|
||||
"""
|
||||
批量查询多个用户的角色列表，按 user_id 分组返回
|
||||
|
||||
用于分页列表等需要一次性获取多个用户角色的场景，
|
||||
避免 N+1 查询问题。
|
||||
|
||||
Args:
|
||||
user_ids: 用户ID列表，如 [1, 2, 3]
|
||||
|
||||
Returns:
|
||||
dict: 按用户ID分组的角色映射，格式：
|
||||
{1: [{"id": 1, "name": "管理员"}, ...], 2: [...]}
|
||||
无数据时返回空字典
|
||||
"""
|
||||
if not user_ids:
|
||||
return {}
|
||||
|
||||
try:
|
||||
result = (
|
||||
db.session.query(UserRole.user_id, Role.id, Role.name)
|
||||
.join(Role, Role.id == UserRole.role_id)
|
||||
.filter(UserRole.user_id.in_(user_ids), Role.is_delete == 0)
|
||||
.all()
|
||||
)
|
||||
|
||||
# 按 user_id 分组
|
||||
role_map = {}
|
||||
for row in result:
|
||||
role_map.setdefault(row.user_id, []).append({'id': row.id, 'name': row.name})
|
||||
return role_map
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"批量查询用户角色列表异常: {str(e)}")
|
||||
return {}
|
||||
|
||||
|
||||
# ============================================================
|
||||
# 角色用户查询
|
||||
# ============================================================
|
||||
|
||||
def get_role_user_list(role_id):
|
||||
"""
|
||||
根据角色ID查询用户列表
|
||||
|
||||
Args:
|
||||
role_id: 角色ID
|
||||
|
||||
Returns:
|
||||
list: 用户列表，包含用户ID和用户名等信息
|
||||
"""
|
||||
if not role_id or role_id <= 0:
|
||||
return []
|
||||
|
||||
try:
|
||||
# ORM JOIN查询，表名自动从Model获取
|
||||
result = (
|
||||
db.session.query(User.id, User.realname, User.username)
|
||||
.join(UserRole, User.id == UserRole.user_id)
|
||||
.filter(UserRole.role_id == role_id, User.is_delete == 0)
|
||||
.all()
|
||||
)
|
||||
|
||||
user_list = [{'id': row.id, 'realname': row.realname, 'username': row.username} for row in result]
|
||||
|
||||
return user_list
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"查询角色用户列表异常，角色ID: {role_id}, 错误: {str(e)}")
|
||||
return []
|
||||
from src.modules.association.user_role.service import * # noqa: F401,F403
|
||||
|
||||
Reference in New Issue
Block a user