v6ole
f1f8518985
feat(auth): 添加用户权限获取接口并完善JWT令牌角色信息 - 在JWT令牌中添加用户角色信息 - 新增get_my_permissions接口用于获取当前用户权限码列表 - 重构认证回调逻辑，增加错误日志记录 - 更新用户信息获取接口使用Authorization头验证 ```
111 lines
3.4 KiB
Python
111 lines
3.4 KiB
Python
"""角色权限配置 API"""
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import text
|
|
from pydantic import BaseModel
|
|
|
|
from app.core.database import get_db
|
|
from app.middleware.permission_middleware import require_permission, invalidate_role_cache
|
|
from app.models.permission import Permission
|
|
|
|
router = APIRouter(prefix="/api", tags=["角色权限"])
|
|
|
|
VALID_ROLES = ['admin', 'area_admin', 'school_admin', 'user']
|
|
|
|
|
|
class RolePermissionsUpdate(BaseModel):
|
|
permissions: list[str] # 权限码列表
|
|
|
|
|
|
@router.get("/permissions")
|
|
def get_all_permissions(
|
|
db: Session = Depends(get_db),
|
|
_: dict = Depends(require_permission('user.view')),
|
|
):
|
|
"""获取所有权限码列表（按模块分组）"""
|
|
perms = db.query(Permission).order_by(Permission.module, Permission.code).all()
|
|
result = {}
|
|
for p in perms:
|
|
module = p.module or 'other'
|
|
if module not in result:
|
|
result[module] = []
|
|
result[module].append({
|
|
"id": p.id,
|
|
"name": p.name,
|
|
"code": p.code,
|
|
"description": p.description,
|
|
})
|
|
return result
|
|
|
|
|
|
@router.get("/roles")
|
|
def get_roles(
|
|
db: Session = Depends(get_db),
|
|
_: dict = Depends(require_permission('user.view')),
|
|
):
|
|
"""获取所有角色及其当前权限"""
|
|
rows = db.execute(
|
|
text("""
|
|
SELECT rp.role, p.code
|
|
FROM role_permissions rp
|
|
JOIN permissions p ON p.id = rp.permission_id
|
|
ORDER BY rp.role, p.code
|
|
""")
|
|
).fetchall()
|
|
|
|
role_map: dict[str, list[str]] = {r: [] for r in VALID_ROLES}
|
|
for role, code in rows:
|
|
if role in role_map:
|
|
role_map[role].append(code)
|
|
|
|
# admin 特殊处理
|
|
role_map['admin'] = ['*']
|
|
|
|
return [
|
|
{"role": role, "permissions": perms}
|
|
for role, perms in role_map.items()
|
|
]
|
|
|
|
|
|
@router.put("/roles/{role}/permissions")
|
|
def update_role_permissions(
|
|
role: str,
|
|
body: RolePermissionsUpdate,
|
|
db: Session = Depends(get_db),
|
|
_: dict = Depends(require_permission('user.manage')),
|
|
):
|
|
"""更新角色权限（替换全量）"""
|
|
if role not in VALID_ROLES:
|
|
raise HTTPException(status_code=400, detail=f"无效角色，可选：{', '.join(VALID_ROLES)}")
|
|
if role == 'admin':
|
|
raise HTTPException(status_code=400, detail="admin 角色权限不可修改")
|
|
|
|
# 验证权限码是否存在
|
|
if body.permissions:
|
|
existing = {p.code for p in db.query(Permission).filter(
|
|
Permission.code.in_(body.permissions)
|
|
).all()}
|
|
invalid = set(body.permissions) - existing
|
|
if invalid:
|
|
raise HTTPException(status_code=400, detail=f"无效权限码：{', '.join(invalid)}")
|
|
|
|
# 删除旧权限，插入新权限
|
|
db.execute(text("DELETE FROM role_permissions WHERE role = :role"), {"role": role})
|
|
if body.permissions:
|
|
# 查出权限 id 再插入，避免 ANY 语法兼容问题
|
|
perm_ids = db.execute(
|
|
text("SELECT id FROM permissions WHERE code IN :codes"),
|
|
{"codes": tuple(body.permissions)}
|
|
).fetchall()
|
|
for (pid,) in perm_ids:
|
|
db.execute(
|
|
text("INSERT INTO role_permissions (role, permission_id) VALUES (:role, :pid)"),
|
|
{"role": role, "pid": pid}
|
|
)
|
|
db.commit()
|
|
|
|
# 清除 Redis 缓存
|
|
invalidate_role_cache(role)
|
|
|
|
return {"message": "权限更新成功"}
|