sonder/fastapi-project-template
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 部门管理添加按钮级权限控制，移除部门角色表

Browse Source
This commit is contained in:
皓月归尘
2025-02-22 04:22:58 +08:00
parent 0c97feade2
commit a804732d53
6 changed files with 90 additions and 368 deletions
Download Patch File Download Diff File Expand all files Collapse all files

279
api/department.py
View File

@@ -10,20 +10,22 @@ from typing import Optional
from fastapi import APIRouter, Depends, Query, Path, Request
from fastapi.responses import JSONResponse
from annotation.auth import Auth
from annotation.log import Log
from config.constant import BusinessType
from config.constant import BusinessType, RedisKeyConfig
from controller.login import LoginController
from models import Department, Role, DepartmentRole
from schemas.common import BaseResponse
from models import Department
from schemas.common import BaseResponse, DeleteListParams
from schemas.department import AddDepartmentParams, GetDepartmentInfoResponse, \
GetDepartmentListResponse, AddDepartmentRoleParams, GetDepartmentRoleInfoResponse, DeleteDepartmentListParams
GetDepartmentListResponse
from utils.response import Response
departmentAPI = APIRouter(prefix="/department", dependencies=[Depends(LoginController.get_current_user)])
departmentAPI = APIRouter(prefix="/department")
@departmentAPI.post("/add", response_model=BaseResponse, response_class=JSONResponse, summary="新增部门")
@Log(title="新增部门", business_type=BusinessType.INSERT)
@Auth(["department:btn:add"])
async def add_department(request: Request, params: AddDepartmentParams,
current_user: dict = Depends(LoginController.get_current_user)):
parent_id = current_user.get("department_id")
@@ -40,6 +42,8 @@ async def add_department(request: Request, params: AddDepartmentParams,
status=params.status
)
if department:
if await request.app.state.redis.get(f'{RedisKeyConfig.USER_INFO.key}:*'):
await request.app.state.redis.delete(f'{RedisKeyConfig.USER_INFO.key}:*')
return Response.success(msg="添加成功！")
else:
return Response.error(msg="添加失败！")
@@ -48,9 +52,16 @@ async def add_department(request: Request, params: AddDepartmentParams,
@departmentAPI.delete("/delete/{id}", response_model=BaseResponse, response_class=JSONResponse, summary="删除部门")
@departmentAPI.post("/delete/{id}", response_model=BaseResponse, response_class=JSONResponse, summary="删除部门")
@Log(title="删除部门", business_type=BusinessType.DELETE)
async def delete_department(request: Request, id: str = Path(description="部门ID")):
@Auth(["department:btn:delete"])
async def delete_department(request: Request, id: str = Path(description="部门ID"),
current_user: dict = Depends(LoginController.get_current_user)):
if department := await Department.get_or_none(id=id, del_flag=1):
sub_departments = current_user.get("sub_departments")
if department.id not in sub_departments:
return Response.error(msg="删除失败,无权限！")
if await delete_department_recursive(department_id=department.id):
if await request.app.state.redis.get(f'{RedisKeyConfig.USER_INFO.key}:*'):
await request.app.state.redis.delete(f'{RedisKeyConfig.USER_INFO.key}:*')
return Response.success(msg="删除成功！")
return Response.error(msg="删除失败!")
else:
@@ -60,10 +71,16 @@ async def delete_department(request: Request, id: str = Path(description="部门
@departmentAPI.delete("/deleteList", response_model=BaseResponse, response_class=JSONResponse, summary="批量删除部门")
@departmentAPI.post("/deleteList", response_model=BaseResponse, response_class=JSONResponse, summary="批量删除部门")
@Log(title="批量删除部门", business_type=BusinessType.DELETE)
async def delete_department_list(request: Request, params: DeleteDepartmentListParams):
@Auth(["department:btn:delete"])
async def delete_department_list(request: Request, params: DeleteListParams,
current_user: dict = Depends(LoginController.get_current_user)):
sub_departments = current_user.get("sub_departments")
for item in set(params.ids):
if department := await Department.get_or_none(id=item, del_flag=1):
if item in sub_departments:
await delete_department_recursive(department_id=department.id)
if await request.app.state.redis.get(f'{RedisKeyConfig.USER_INFO.key}:*'):
await request.app.state.redis.delete(f'{RedisKeyConfig.USER_INFO.key}:*')
return Response.success(msg="删除成功！")
@@ -73,8 +90,8 @@ async def delete_department_recursive(department_id: str):
:param department_id: 部门ID
:return:
"""
await Department.filter(id=department_id).delete()
sub_departments = await Department.filter(parent_id=department_id).all()
await Department.filter(id=department_id, del_flag=1).update(del_flag=0)
sub_departments = await Department.filter(parent_id=department_id, del_flag=1).all()
for sub_department in sub_departments:
await delete_department_recursive(sub_department.id)
return True
@@ -83,8 +100,13 @@ async def delete_department_recursive(department_id: str):
@departmentAPI.put("/update/{id}", response_model=BaseResponse, response_class=JSONResponse, summary="修改部门")
@departmentAPI.post("/update/{id}", response_model=BaseResponse, response_class=JSONResponse, summary="修改部门")
@Log(title="修改部门", business_type=BusinessType.UPDATE)
async def update_department(request: Request, params: AddDepartmentParams, id: str = Path(description="部门ID")):
@Auth(["department:btn:update"])
async def update_department(request: Request, params: AddDepartmentParams, id: str = Path(description="部门ID"),
current_user: dict = Depends(LoginController.get_current_user)):
if department := await Department.get_or_none(id=id, del_flag=1):
sub_departments = current_user.get("sub_departments")
if id not in sub_departments:
return Response.error(msg="修改失败,无权限！")
department.name = params.name
department.parent_id = params.parent_id
department.principal = params.principal
@@ -102,7 +124,9 @@ async def update_department(request: Request, params: AddDepartmentParams, id: s
@departmentAPI.get("/info/{id}", response_model=GetDepartmentInfoResponse, response_class=JSONResponse,
summary="查询部门详情")
@Log(title="查询部门详情", business_type=BusinessType.SELECT)
async def get_department(request: Request, id: str = Path(description="部门ID")):
@Auth(["department:btn:info"])
async def get_department(request: Request, id: str = Path(description="部门ID"),
current_user: dict = Depends(LoginController.get_current_user)):
if department := await Department.get_or_none(id=id, del_flag=1).values(
id="id",
name="name",
@@ -118,6 +142,9 @@ async def get_department(request: Request, id: str = Path(description="部门ID"
create_by="create_by",
update_by="update_by"
):
sub_departments = current_user.get("sub_departments")
if id not in sub_departments:
return Response.error(msg="查询失败,无权限！")
return Response.success(data=department)
else:
return Response.error(msg="部门不存在！")
@@ -126,6 +153,7 @@ async def get_department(request: Request, id: str = Path(description="部门ID"
@departmentAPI.get("/list", response_model=GetDepartmentListResponse, response_class=JSONResponse,
summary="查询部门列表")
@Log(title="查询部门列表", business_type=BusinessType.SELECT)
@Auth(["department:btn:list"])
async def get_department_list(
request: Request,
page: int = Query(default=1, description="当前页码"),
@@ -148,43 +176,11 @@ async def get_department_list(
'sort': sort
}.items() if v
}
department_id = current_user.get("department_id", "")
# 递归查询所有部门
all_departments = await get_department_and_subdepartments(department_id, filterArgs)
# 分页处理
total = len(all_departments)
paginated_departments = all_departments[(page - 1) * pageSize: page * pageSize]
return Response.success(data={
"result": paginated_departments,
"total": total,
"page": page
})
async def get_department_and_subdepartments(department_id: str, filterArgs: dict, visited: set = None):
"""
查询当前部门及其所有下属部门的数据，并根据 id 去重。
:param department_id: 当前部门 ID
:param filterArgs: 过滤条件
:param visited: 已访问的部门 ID 集合，用于避免循环依赖
:return: 去重后的部门列表
"""
if visited is None:
visited = set() # 初始化已访问的部门 ID 集合
# 如果当前部门 ID 已经访问过，直接返回空列表，避免死循环
if department_id in visited:
return []
visited.add(department_id) # 标记当前部门 ID 为已访问
# 查询当前部门
current_department = await Department.filter(
id=department_id,
**filterArgs
).values(
sub_departments = current_user.get("sub_departments")
total = await Department.filter(**filterArgs, del_flag=1, id__in=sub_departments).count()
data = await Department.filter(**filterArgs, del_flag=1, id__in=sub_departments).offset(
(page - 1) * pageSize).limit(
pageSize).values(
id="id",
name="name",
parent_id="parent_id",
@@ -199,192 +195,9 @@ async def get_department_and_subdepartments(department_id: str, filterArgs: dict
create_by="create_by",
update_by="update_by"
)
# 查询直接子部门
sub_departments = await Department.filter(
parent_id=department_id, # 只根据 parent_id 查询
**filterArgs
).values(
id="id",
name="name",
parent_id="parent_id",
principal="principal",
phone="phone",
email="email",
remark="remark",
sort="sort",
status="status",
create_time="create_time",
update_time="update_time",
create_by="create_by",
update_by="update_by"
)
# 递归查询子部门的子部门
for department in sub_departments[:]: # 使用切片复制避免修改迭代中的列表
sub_sub_departments = await get_department_and_subdepartments(department["id"], filterArgs, visited)
sub_departments.extend(sub_sub_departments)
# 合并当前部门和所有下属部门的数据
all_departments = current_department + sub_departments
# 根据 id 去重
unique_departments = []
seen_ids = set() # 用于记录已经处理过的部门 ID
for department in all_departments:
if department["id"] not in seen_ids:
unique_departments.append(department)
seen_ids.add(department["id"])
return unique_departments
@departmentAPI.post("/addRole", response_model=BaseResponse, response_class=JSONResponse, summary="添加部门角色")
@Log(title="添加部门角色", business_type=BusinessType.INSERT)
async def add_department_role(request: Request, params: AddDepartmentRoleParams):
if await DepartmentRole.get_or_none(department_id=params.department_id, role_id=params.role_id, del_flag=1):
return Response.error(msg="该部门已存在该角色！")
if department := await Department.get_or_none(id=params.department_id, del_flag=1):
if role := await Role.get_or_none(id=params.role_id, del_flag=1):
departmentRole = await DepartmentRole.create(department_id=department.id, role_id=role.id)
if departmentRole:
return Response.success(msg="添加成功！")
else:
return Response.error(msg="添加失败！")
else:
return Response.error(msg="添加失败,角色不存在！")
else:
return Response.error(msg="添加失败,部门不存在！")
@departmentAPI.delete("/deleteRole/{id}", response_model=BaseResponse, response_class=JSONResponse,
summary="删除部门角色")
@departmentAPI.post("/deleteRole/{id}", response_model=BaseResponse, response_class=JSONResponse,
summary="删除部门角色")
@Log(title="删除部门角色", business_type=BusinessType.DELETE)
async def delete_department_role(request: Request, id: str = Path(description="部门角色ID")):
if departmentRole := await DepartmentRole.get_or_none(id=id, del_flag=1):
await departmentRole.delete()
return Response.success(msg="删除成功！")
else:
return Response.error(msg="删除失败,部门角色不存在！")
@departmentAPI.put("/updateRole/{id}", response_model=BaseResponse, response_class=JSONResponse, summary="修改部门角色")
@departmentAPI.post("/updateRole/{id}", response_model=BaseResponse, response_class=JSONResponse,
summary="修改部门角色")
@Log(title="修改部门角色", business_type=BusinessType.UPDATE)
async def update_department_role(request: Request, params: AddDepartmentRoleParams,
id: str = Path(description="部门角色ID")):
if departmentRole := await DepartmentRole.get_or_none(id=id, del_flag=1):
if department := await Department.get_or_none(id=params.department_id, del_flag=1):
if role := await Role.get_or_none(id=params.role_id, del_flag=1):
departmentRole.department_id = department.id
departmentRole.role_id = role.id
await departmentRole.save()
return Response.success(msg="修改成功！")
else:
return Response.error(msg="修改失败,角色不存在！")
else:
return Response.error(msg="修改失败,部门不存在！")
else:
return Response.error(msg="修改失败,部门角色不存在！")
@departmentAPI.get("/roleInfo", response_model=GetDepartmentRoleInfoResponse, response_class=JSONResponse,
summary="获取部门角色信息")
@Log(title="获取部门角色信息", business_type=BusinessType.SELECT)
async def get_department_role_info(request: Request, id: str = Query(description="部门角色ID")):
if departmentRole := await DepartmentRole.get_or_none(id=id, del_flag=1):
data = await departmentRole.first().values(
id="id",
department_id="department__id",
department_name="department__name",
department_phone="department__phone",
department_principal="department__principal",
department_email="department__email",
role_name="role__name",
role_code="role__code",
role_id="role__id",
create_time="create_time",
update_time="update_time"
)
return Response.success(data=data)
else:
return Response.error(msg="获取失败,部门角色不存在！")
@departmentAPI.get("/roleList", response_model=GetDepartmentListResponse, response_class=JSONResponse,
summary="获取部门角色列表")
@Log(title="获取部门角色列表", business_type=BusinessType.SELECT)
async def get_department_role_list(
request: Request,
page: int = Query(default=1, description="当前页码"),
pageSize: int = Query(default=10, description="每页条数"),
department_id: Optional[str] = Query(default=None, description="部门ID"),
department_name: Optional[str] = Query(default=None, description="部门名称"),
department_phone: Optional[str] = Query(default=None, description="部门电话"),
department_principal: Optional[str] = Query(default=None, description="部门负责人"),
department_email: Optional[str] = Query(default=None, description="部门邮箱"),
role_id: Optional[str] = Query(default=None, description="角色ID"),
role_name: Optional[str] = Query(default=None, description="角色名称"),
role_code: Optional[str] = Query(default=None, description="角色编码"),
):
filterArgs = {
f'{k}__contains': v for k, v in {
'department__id': department_id,
'department__name': department_name,
'department__phone': department_phone,
'department__principal': department_principal,
'department__email': department_email,
'role__id': role_id,
'role__name': role_name,
'role__code': role_code
}.items() if v
}
total = await DepartmentRole.filter(**filterArgs).count()
data = await DepartmentRole.filter(**filterArgs).offset((page - 1) * pageSize).limit(pageSize).values(
id="id",
department_id="department__id",
department_name="department__name",
department_phone="department__phone",
department_principal="department__principal",
department_email="department__email",
role_name="role__name",
role_code="role__code",
role_id="role__id",
create_time="create_time",
update_time="update_time"
)
return Response.success(data={
"result": data,
"total": total,
"page": page
})
@departmentAPI.get("/roleList/{id}", response_model=GetDepartmentListResponse, response_class=JSONResponse,
summary="用户获取部门角色列表")
@Log(title="获取部门角色列表", business_type=BusinessType.OTHER)
async def get_department_role_list(
request: Request,
id: str = Path(..., description="部门ID")
):
data = await Role.filter(department__id=id).values(
id="id",
department_id="department__id",
department_name="department__name",
department_phone="department__phone",
department_principal="department__principal",
department_email="department__email",
role_name="name",
role_code="code",
role_id="id",
create_time="create_time",
update_time="update_time"
)
return Response.success(data={
"result": data,
"total": len(data),
"page": 1
"page": page,
"pageSize": pageSize
})

52
api/user.py
View File

@@ -22,7 +22,7 @@ from exceptions.exception import ModelValidatorException
from models import File as FileModel
from models import Role, Department
from models.user import User, UserRole
from schemas.common import BaseResponse
from schemas.common import BaseResponse, DeleteListParams
from schemas.department import GetDepartmentListResponse
from schemas.file import UploadFileResponse
from schemas.user import AddUserParams, GetUserListResponse, GetUserInfoResponse, UpdateUserParams, \
@@ -46,7 +46,7 @@ async def add_user(
if await QueryController.register_user_before(username=params.username, phone=params.phone, email=params.email):
return Response.error(msg="添加失败，用户已存在！")
params.password = await Password.get_password_hash(input_password=params.password)
department = await Department.get_or_none(id=params.department_id)
department = await Department.get_or_none(id=params.department_id, del_flag=1)
user = await User.create(
username=params.username,
password=params.password,
@@ -73,13 +73,31 @@ async def delete_user(
current_user: dict = Depends(LoginController.get_current_user)
):
sub_departments = current_user.get("sub_departments")
if user := await User.get_or_none(id=id, department__id__in=sub_departments):
await user.delete()
if user := await User.get_or_none(id=id, department__id__in=sub_departments, del_flag=1):
user.del_flag = 0
await user.save()
return Response.success(msg="删除成功！")
else:
return Response.error(msg="删除失败，用户不存在！")
@userAPI.delete("/deleteUserList", response_class=JSONResponse, response_model=BaseResponse, summary="批量删除用户")
@userAPI.post("/deleteUserList", response_class=JSONResponse, response_model=BaseResponse, summary="批量删除用户")
@Log(title="批量删除用户", business_type=BusinessType.DELETE)
@Auth(["user:btn:deleteUser"])
async def delete_user_list(
request: Request,
params: DeleteListParams,
current_user: dict = Depends(LoginController.get_current_user)
):
sub_departments = current_user.get("sub_departments")
for id in params.ids:
if user := await User.get_or_none(id=id, department__id__in=sub_departments,del_flag=1):
user.del_flag=0
await user.save()
return Response.success(msg="删除成功！")
@userAPI.put("/update/{id}", response_class=JSONResponse, response_model=BaseResponse, summary="更新用户")
@userAPI.post("/update/{id}", response_class=JSONResponse, response_model=BaseResponse, summary="更新用户")
@Log(title="更新用户", business_type=BusinessType.UPDATE)
@@ -91,14 +109,14 @@ async def update_user(
current_user: dict = Depends(LoginController.get_current_user)
):
sub_departments = current_user.get("sub_departments")
if user := await User.get_or_none(id=id, department__id__in=sub_departments):
if user := await User.get_or_none(id=id, department__id__in=sub_departments,del_flag=1):
user.username = params.username
user.nickname = params.nickname
user.phone = params.phone
user.email = params.email
user.gender = params.gender
user.status = params.status
if department := await Department.get_or_none(id=params.department_id):
if department := await Department.get_or_none(id=params.department_id,del_flag=1):
user.department = department
else:
user.department = None
@@ -115,7 +133,7 @@ async def update_user(
@Auth(["user:btn:Userinfo"])
async def get_user_info(request: Request, id: str = Path(..., description="用户ID"),
current_user: dict = Depends(LoginController.get_current_user)):
if user := await User.get_or_none(id=id):
if user := await User.get_or_none(id=id,del_flag=1):
user = await user.first().values(
id="id",
create_time="create_time",
@@ -164,8 +182,8 @@ async def get_user_list(
}
if not department_id:
filterArgs['department_id__in'] = sub_departments
total = await User.filter(**filterArgs).count()
result = await User.filter(**filterArgs).offset((page - 1) * pageSize).limit(pageSize).values(
total = await User.filter(**filterArgs,del_flag=1).count()
result = await User.filter(**filterArgs,del_flag=1).offset((page - 1) * pageSize).limit(pageSize).values(
id="id",
create_time="create_time",
update_time="update_time",
@@ -181,7 +199,8 @@ async def get_user_list(
return Response.success(data={
"result": result,
"total": total,
"page": page
"page": page,
"pageSize": pageSize
})
@@ -219,7 +238,8 @@ async def delete_user_role(request: Request, id: str = Path(description="用户
current_user: dict = Depends(LoginController.get_current_user)):
sub_departments = current_user.get("sub_departments")
if userRole := await UserRole.get_or_none(id=id, del_flag=1, user__department__id__in=sub_departments):
await userRole.delete()
userRole.del_flag=0
await userRole.save()
if await request.app.state.redis.get(f'{RedisKeyConfig.USER_INFO.key}:{current_user.get("id")}'):
await request.app.state.redis.delete(f'{RedisKeyConfig.USER_INFO.key}:{current_user.get("id")}')
return Response.success(msg="删除成功！")
@@ -251,7 +271,8 @@ async def update_user_role(request: Request, params: UpdateUserRoleParams,
for role_id in deleteRoles:
if userRole := await UserRole.get_or_none(user_id=params.user_id, role_id=role_id, del_flag=1,
user__department__id__in=sub_departments):
await userRole.delete()
userRole.del_flag=0
await userRole.save()
if await request.app.state.redis.get(f'{RedisKeyConfig.USER_INFO.key}:{params.user_id}'):
await request.app.state.redis.delete(f'{RedisKeyConfig.USER_INFO.key}:{params.user_id}')
return Response.success(msg="修改成功！")
@@ -306,7 +327,8 @@ async def get_user_role_list(
return Response.success(data={
"result": result,
"total": len(result),
"page": 1
"page": 1,
"pageSize": 10,
})
@@ -331,7 +353,7 @@ async def upload_user_avatar(
id: str = Path(description="用户ID"),
file: UploadFile = File(...), current_user: dict = Depends(LoginController.get_current_user)):
sub_departments = current_user.get("sub_departments")
if user := await User.get_or_none(id=id, department__id__in=sub_departments):
if user := await User.get_or_none(id=id, del_flag=1,department__id__in=sub_departments):
image_mimetypes = [
'image/jpeg',
'image/png',
@@ -395,7 +417,7 @@ async def upload_user_avatar(
@Auth(permission_list=["user:btn:reset_password"])
async def reset_user_password(request: Request, params: ResetPasswordParams, id: str = Path(description="用户ID"),
current_user: dict = Depends(LoginController.get_current_user)):
if user := await User.get_or_none(id=id, department__id__in=current_user.get("sub_departments")):
if user := await User.get_or_none(id=id,del_flag=1, department__id__in=current_user.get("sub_departments")):
user.password = await Password.get_password_hash(params.password)
await user.save()
return Response.success(msg="重置密码成功！")

5
controller/query.py
View File

@@ -8,7 +8,7 @@
from typing import Union
from tortoise.expressions import Q
from fastapi import Request
from models import User, UserRole, RolePermission, Department
from utils.common import filterKeyValues
@@ -141,11 +141,12 @@ class QueryController:
async def get_sub_department_ids(cls, department_id: str) -> list:
# 递归获取指定部门及其所有下属部门的 ID
async def fetch_sub_deps(dep_id: str):
sub_deps = await Department.filter(parent_id=dep_id).all()
sub_deps = await Department.filter(parent_id=dep_id, del_flag=1).all()
sub_deps_list = [dep.id for dep in sub_deps]
for sub_dep in sub_deps:
sub_deps_list.extend(await fetch_sub_deps(sub_dep.id)) # 递归获取下属部门
return sub_deps_list
dataList = await fetch_sub_deps(department_id)
dataList.append(department_id)
return list(set(dataList))

3
models/__init__.py
View File

@@ -7,7 +7,7 @@
# @Comment : 本程序
from models.config import Config
from models.department import Department, DepartmentRole
from models.department import Department
from models.file import File
from models.i18n import I18n, Locale
from models.log import LoginLog, OperationLog
@@ -17,7 +17,6 @@ from models.user import User, UserRole
__all__ = [
'Department',
'DepartmentRole',
'File',
'LoginLog',
'OperationLog',

35
models/department.py
View File

@@ -118,38 +118,3 @@ class Department(BaseModel):
table = "department" # 数据库表名
table_description = "部门表" # 表描述
ordering = ["sort", "-create_time"] # 默认按排序权重和创建时间排序
class DepartmentRole(BaseModel):
"""
部门角色表模型。
"""
department = fields.ForeignKeyField(
"models.Department",
related_name="department_roles",
description="部门ID",
source_field="department_id" # 映射到数据库字段 department_id
)
"""
部门ID。
- 外键关联到 Department 表。
- 映射到数据库字段 department_id。
"""
role = fields.ForeignKeyField(
"models.Role",
related_name="department_roles",
description="角色ID",
source_field="role_id" # 映射到数据库字段 role_id
)
"""
角色ID。
- 外键关联到 Role 表。
- 映射到数据库字段 role_id。
"""
class Meta:
table = "department_role" # 数据库表名
table_description = "部门角色表" # 表描述
unique_together = (("department_id", "role_id"),) # 唯一约束，防止重复分配

78
schemas/department.py
View File

@@ -86,13 +86,6 @@ class AddDepartmentParams(BaseModel):
}
class DeleteDepartmentListParams(BaseModel):
"""
删除部门参数模型。
"""
ids: List[str] = Field(..., description="部门ID列表")
class GetDepartmentListResult(ListQueryResult):
"""
获取部门列表结果模型。
@@ -105,74 +98,3 @@ class GetDepartmentListResponse(BaseResponse):
获取部门列表响应模型。
"""
data: GetDepartmentListResult = Field(default=None, description="响应数据")
class AddDepartmentRoleParams(BaseModel):
"""
添加部门角色参数模型。
"""
department_id: str = Field(..., max_length=36, description="部门ID")
role_id: str = Field(..., max_length=36, description="角色ID")
class Config:
json_schema_extra = {
"example": {
"department_id": "550e8400-e29b-41d4-a716-446655440000",
"role_id": "550e8400-e29b-41d4-a716-446655440000"
}
}
class DepartmentRoleInfo(BaseModel):
"""
部门角色信息模型。
"""
id: str = Field(..., max_length=36, description="主键ID")
department_id: str = Field(..., max_length=36, description="部门ID")
department_name: str = Field(..., max_length=100, description="部门名称")
department_phone: str = Field(..., max_length=30, description="部门电话")
department_principal: str = Field(..., max_length=64, description="部门负责人")
department_email: str = Field(..., max_length=128, description="部门邮箱")
role_name: str = Field(..., max_length=100, description="角色名称")
role_code: str = Field(..., max_length=100, description="角色编码")
role_id: str = Field(..., max_length=36, description="角色ID")
create_time: datetime = Field(..., description="创建时间")
update_time: datetime = Field(..., description="更新时间")
class Config:
json_schema_extra = {
"example": {
"id": "550e8400-e29b-41d4-a716-446655440000",
"department_id": "550e8400-e29b-41d4-a716-446655440000",
"department_name": "研发部",
"department_phone": "1234567890",
"department_principal": "张三",
"department_email": "dev@example.com",
"role_name": "管理员",
"role_code": "admin",
"role_id": "550e8400-e29b-41d4-a716-446655440000",
"create_time": "2023-10-01T12:00:00",
"update_time": "2023-10-01T12:00:00"
}
}
class GetDepartmentRoleInfoResponse(BaseResponse):
"""
获取部门角色信息响应模型。
"""
data: DepartmentRoleInfo = Field(default=None, description="响应数据")
class GetDepartmentRoleListResult(ListQueryResult):
"""
获取部门角色列表结果模型。
"""
result: List[DepartmentRoleInfo] = Field(default=[], description="部门角色列表")
class GetDepartmentRoleListResponse(BaseResponse):
"""
获取部门角色列表响应模型。
"""
data: GetDepartmentRoleListResult = Field(default=None, description="响应数据")