686 lines
23 KiB
Python
686 lines
23 KiB
Python
from collections.abc import Callable
|
|
from datetime import datetime, timezone
|
|
from typing import NoReturn
|
|
|
|
from fastapi import HTTPException, status
|
|
from sqlalchemy import delete, func, select
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.field_crypto import encrypt_text
|
|
from app.core.masking import mask_mobile
|
|
from app.core.security import create_access_token
|
|
from app.models.rbac import (
|
|
AdminResource,
|
|
AdminRole,
|
|
AdminRoleResource,
|
|
AdminUser,
|
|
AdminUserRole,
|
|
utcnow,
|
|
)
|
|
from app.services.captcha import consume_captcha
|
|
from app.services.login_guard import clear_failures, is_currently_locked, record_failure
|
|
from app.services.password import hash_password, validate_password, verify_password
|
|
|
|
LOGIN_FAILURE = "用户名或密码错误"
|
|
KEEP_FULL_ADMIN = "至少保留一名拥有全部系统菜单的正常用户"
|
|
KEEP_SELF = "不能停用或注销当前登录用户"
|
|
|
|
USER_STATUSES = {"active", "disabled", "cancelled"}
|
|
ROLE_STATUSES = {"active", "disabled"}
|
|
RESOURCE_STATUSES = {"active", "disabled"}
|
|
RESOURCE_TYPES = {"directory", "menu", "button"}
|
|
MENU_TYPES = {"directory", "menu"}
|
|
|
|
|
|
def isoformat(value: datetime | None) -> str | None:
|
|
if value is None:
|
|
return None
|
|
if value.tzinfo is None:
|
|
value = value.replace(tzinfo=timezone.utc)
|
|
return value.isoformat()
|
|
|
|
|
|
def _bad(detail: str, code: int = status.HTTP_400_BAD_REQUEST) -> NoReturn:
|
|
raise HTTPException(status_code=code, detail=detail)
|
|
|
|
|
|
def _require_text(value: str | None, *, empty: str, limit: int, too_long: str) -> str:
|
|
text = (value or "").strip()
|
|
if not text:
|
|
_bad(empty)
|
|
if len(text) > limit:
|
|
_bad(too_long)
|
|
return text
|
|
|
|
|
|
def _optional_text(value: str | None, limit: int, too_long: str) -> str | None:
|
|
if value is None:
|
|
return None
|
|
text = value.strip()
|
|
if not text:
|
|
return None
|
|
if len(text) > limit:
|
|
_bad(too_long)
|
|
return text
|
|
|
|
|
|
def _clean(value: str | None, limit: int, too_long: str) -> str:
|
|
text = (value or "").strip()
|
|
if len(text) > limit:
|
|
_bad(too_long)
|
|
return text
|
|
|
|
|
|
def _exact(value: str | None) -> str | None:
|
|
if value is None:
|
|
return None
|
|
text = value.strip()
|
|
return text or None
|
|
|
|
|
|
def _page(current: int, size: int) -> tuple[int, int]:
|
|
if current < 1:
|
|
current = 1
|
|
if size < 1:
|
|
size = 10
|
|
if size > 200:
|
|
size = 200
|
|
return current, size
|
|
|
|
|
|
def commit_or_raise(db: Session, message: str) -> None:
|
|
try:
|
|
db.commit()
|
|
except IntegrityError:
|
|
db.rollback()
|
|
_bad(message)
|
|
|
|
|
|
def preserve_full_admin(db: Session, mutate: Callable[[], None]) -> None:
|
|
before = count_full_admins(db)
|
|
with db.begin_nested():
|
|
mutate()
|
|
db.flush()
|
|
if before >= 1 and count_full_admins(db) < 1:
|
|
_bad(KEEP_FULL_ADMIN)
|
|
|
|
|
|
def count_full_admins(db: Session) -> int:
|
|
needed = set(db.scalars(select(AdminResource.id).where(AdminResource.status == "active")).all())
|
|
user_ids = db.scalars(select(AdminUser.id).where(AdminUser.status == "active")).all()
|
|
total = 0
|
|
for user_id in user_ids:
|
|
if needed <= owned_resource_ids(db, user_id):
|
|
total += 1
|
|
return total
|
|
|
|
|
|
def owned_resource_ids(db: Session, user_id: int) -> set[int]:
|
|
rows = db.scalars(
|
|
select(AdminRoleResource.resource_id)
|
|
.join(AdminRole, AdminRole.id == AdminRoleResource.role_id)
|
|
.join(AdminUserRole, AdminUserRole.role_id == AdminRole.id)
|
|
.where(AdminUserRole.user_id == user_id, AdminRole.status == "active")
|
|
).all()
|
|
return set(rows)
|
|
|
|
|
|
def apply_mobile(user: AdminUser, mobile: str) -> None:
|
|
text = mobile.strip()
|
|
if len(text) > 20:
|
|
_bad("手机号过长")
|
|
if not text:
|
|
user.mobile_cipher = None
|
|
user.mobile_mask = ""
|
|
return
|
|
user.mobile_cipher = encrypt_text(text)
|
|
user.mobile_mask = mask_mobile(text)
|
|
|
|
|
|
def ensure_role_ids(db: Session, role_ids: list[int]) -> list[int]:
|
|
unique = list(dict.fromkeys(role_ids))
|
|
if not unique:
|
|
return []
|
|
found = set(db.scalars(select(AdminRole.id).where(AdminRole.id.in_(unique))).all())
|
|
if found != set(unique):
|
|
_bad("角色不存在")
|
|
return unique
|
|
|
|
|
|
def ensure_resource_ids(db: Session, resource_ids: list[int]) -> list[int]:
|
|
unique = list(dict.fromkeys(resource_ids))
|
|
if not unique:
|
|
return []
|
|
found = set(db.scalars(select(AdminResource.id).where(AdminResource.id.in_(unique))).all())
|
|
if found != set(unique):
|
|
_bad("资源不存在")
|
|
return unique
|
|
|
|
|
|
def replace_user_roles(db: Session, user_id: int, role_ids: list[int]) -> None:
|
|
db.execute(delete(AdminUserRole).where(AdminUserRole.user_id == user_id))
|
|
for role_id in role_ids:
|
|
db.add(AdminUserRole(user_id=user_id, role_id=role_id))
|
|
|
|
|
|
def replace_role_links(db: Session, role_id: int, resource_ids: list[int]) -> None:
|
|
db.execute(delete(AdminRoleResource).where(AdminRoleResource.role_id == role_id))
|
|
for resource_id in resource_ids:
|
|
db.add(AdminRoleResource(role_id=role_id, resource_id=resource_id))
|
|
|
|
|
|
def _role_map(db: Session, user_ids: list[int]) -> dict[int, list[AdminRole]]:
|
|
grouped: dict[int, list[AdminRole]] = {user_id: [] for user_id in user_ids}
|
|
if not user_ids:
|
|
return grouped
|
|
rows = db.execute(
|
|
select(AdminUserRole.user_id, AdminRole)
|
|
.join(AdminRole, AdminRole.id == AdminUserRole.role_id)
|
|
.where(AdminUserRole.user_id.in_(user_ids))
|
|
.order_by(AdminRole.id)
|
|
).all()
|
|
for user_id, role in rows:
|
|
grouped.setdefault(user_id, []).append(role)
|
|
return grouped
|
|
|
|
|
|
def _resource_id_map(db: Session, role_ids: list[int]) -> dict[int, list[int]]:
|
|
grouped: dict[int, list[int]] = {role_id: [] for role_id in role_ids}
|
|
if not role_ids:
|
|
return grouped
|
|
rows = db.execute(
|
|
select(AdminRoleResource.role_id, AdminRoleResource.resource_id)
|
|
.where(AdminRoleResource.role_id.in_(role_ids))
|
|
.order_by(AdminRoleResource.resource_id)
|
|
).all()
|
|
for role_id, resource_id in rows:
|
|
grouped.setdefault(role_id, []).append(resource_id)
|
|
return grouped
|
|
|
|
|
|
def serialize_login_user(user: AdminUser) -> dict:
|
|
return {
|
|
"id": user.id,
|
|
"username": user.username,
|
|
"displayName": user.display_name,
|
|
"mobileMask": user.mobile_mask or "",
|
|
"status": user.status,
|
|
}
|
|
|
|
|
|
def serialize_user(user: AdminUser, roles: list[AdminRole]) -> dict:
|
|
return {
|
|
"id": user.id,
|
|
"username": user.username,
|
|
"displayName": user.display_name,
|
|
"mobileMask": user.mobile_mask or "",
|
|
"status": user.status,
|
|
"roleIds": [role.id for role in roles],
|
|
"roles": [{"id": role.id, "code": role.code, "name": role.name} for role in roles],
|
|
"createdAt": isoformat(user.created_at),
|
|
"updatedAt": isoformat(user.updated_at),
|
|
}
|
|
|
|
|
|
def serialize_role(role: AdminRole, resource_ids: list[int]) -> dict:
|
|
return {
|
|
"id": role.id,
|
|
"code": role.code,
|
|
"name": role.name,
|
|
"description": role.description,
|
|
"status": role.status,
|
|
"resourceIds": resource_ids,
|
|
"createdAt": isoformat(role.created_at),
|
|
"updatedAt": isoformat(role.updated_at),
|
|
}
|
|
|
|
|
|
def serialize_node(resource: AdminResource, children: list[dict]) -> dict:
|
|
return {
|
|
"id": resource.id,
|
|
"parentId": resource.parent_id,
|
|
"type": resource.type,
|
|
"name": resource.name,
|
|
"title": resource.title,
|
|
"path": resource.path or "",
|
|
"component": resource.component or "",
|
|
"icon": resource.icon or "",
|
|
"permission": resource.permission or "",
|
|
"sort": resource.sort,
|
|
"visible": bool(resource.visible),
|
|
"status": resource.status,
|
|
"children": children,
|
|
}
|
|
|
|
|
|
def build_tree(resources: list[AdminResource]) -> list[dict]:
|
|
known = {item.id for item in resources}
|
|
grouped: dict[int | None, list[AdminResource]] = {}
|
|
for item in resources:
|
|
parent_id = item.parent_id if item.parent_id in known else None
|
|
grouped.setdefault(parent_id, []).append(item)
|
|
for nodes in grouped.values():
|
|
nodes.sort(key=lambda node: (node.sort, node.id))
|
|
|
|
def walk(node: AdminResource) -> dict:
|
|
children = [walk(child) for child in grouped.get(node.id, [])]
|
|
return serialize_node(node, children)
|
|
|
|
return [walk(node) for node in grouped.get(None, [])]
|
|
|
|
|
|
def sibling_name_taken(db: Session, parent_id: int | None, name: str, exclude_id: int | None = None) -> bool:
|
|
stmt = select(AdminResource.id).where(AdminResource.name == name)
|
|
if parent_id is None:
|
|
stmt = stmt.where(AdminResource.parent_id.is_(None))
|
|
else:
|
|
stmt = stmt.where(AdminResource.parent_id == parent_id)
|
|
if exclude_id is not None:
|
|
stmt = stmt.where(AdminResource.id != exclude_id)
|
|
return db.scalar(stmt) is not None
|
|
|
|
|
|
def assert_parent(db: Session, resource_id: int | None, parent_id: int | None) -> None:
|
|
if parent_id is None:
|
|
return
|
|
if resource_id is not None and parent_id == resource_id:
|
|
_bad("不能把资源移动到自己的下级")
|
|
seen: set[int] = set()
|
|
current: int | None = parent_id
|
|
while current is not None:
|
|
if current in seen:
|
|
break
|
|
if resource_id is not None and current == resource_id:
|
|
_bad("不能把资源移动到自己的下级")
|
|
seen.add(current)
|
|
parent = db.get(AdminResource, current)
|
|
if parent is None:
|
|
_bad("上级资源不存在")
|
|
current = parent.parent_id
|
|
|
|
|
|
def granted_resources(db: Session, user_id: int) -> list[AdminResource]:
|
|
stmt = select(AdminResource).where(
|
|
AdminResource.status == "active",
|
|
AdminResource.id.in_(
|
|
select(AdminRoleResource.resource_id)
|
|
.join(AdminRole, AdminRole.id == AdminRoleResource.role_id)
|
|
.join(AdminUserRole, AdminUserRole.role_id == AdminRole.id)
|
|
.where(AdminUserRole.user_id == user_id, AdminRole.status == "active")
|
|
),
|
|
)
|
|
return list(db.scalars(stmt).all())
|
|
|
|
|
|
def menus_and_permissions(db: Session, user: AdminUser) -> tuple[list[dict], list[str]]:
|
|
resources = granted_resources(db, user.id)
|
|
menus = build_tree([item for item in resources if item.type in MENU_TYPES])
|
|
permissions = sorted({item.permission for item in resources if item.type == "button" and item.permission})
|
|
return menus, permissions
|
|
|
|
|
|
def _reject_login(db: Session, user: AdminUser | None, *, record: bool) -> NoReturn:
|
|
if record and user is not None:
|
|
record_failure(db, user)
|
|
db.commit()
|
|
_bad(LOGIN_FAILURE, status.HTTP_401_UNAUTHORIZED)
|
|
|
|
|
|
def authenticate(db: Session, username: str, password: str, captcha_id: str, captcha_code: str) -> dict:
|
|
captcha_ok = consume_captcha(captcha_id, captcha_code)
|
|
user = db.scalar(select(AdminUser).where(AdminUser.username == (username or "").strip()))
|
|
if user is None:
|
|
_reject_login(db, None, record=False)
|
|
if is_currently_locked(user):
|
|
_reject_login(db, user, record=False)
|
|
if not captcha_ok:
|
|
_reject_login(db, user, record=True)
|
|
if not verify_password(password, user.password_hash):
|
|
_reject_login(db, user, record=True)
|
|
if user.status != "active":
|
|
_reject_login(db, user, record=True)
|
|
clear_failures(db, user)
|
|
token = create_access_token(str(user.id), "staff")
|
|
menus, permissions = menus_and_permissions(db, user)
|
|
db.commit()
|
|
return {
|
|
"token": token,
|
|
"user": serialize_login_user(user),
|
|
"menus": menus,
|
|
"permissions": permissions,
|
|
}
|
|
|
|
|
|
def change_own_password(db: Session, user_id: int, old_password: str, new_password: str) -> dict:
|
|
user = db.get(AdminUser, user_id)
|
|
if user is None:
|
|
_bad("登录已失效", status.HTTP_401_UNAUTHORIZED)
|
|
if not verify_password(old_password, user.password_hash):
|
|
_bad("旧密码错误")
|
|
validate_password(new_password)
|
|
user.password_hash = hash_password(new_password)
|
|
user.updated_at = utcnow()
|
|
commit_or_raise(db, "保存密码失败")
|
|
return {"ok": True}
|
|
|
|
|
|
def query_users(
|
|
db: Session,
|
|
*,
|
|
current: int,
|
|
size: int,
|
|
username: str | None,
|
|
display_name: str | None,
|
|
status_value: str | None,
|
|
) -> dict:
|
|
current, size = _page(current, size)
|
|
filters = []
|
|
username_text = _exact(username)
|
|
display_text = _exact(display_name)
|
|
status_text = _exact(status_value)
|
|
if username_text:
|
|
filters.append(AdminUser.username.contains(username_text))
|
|
if display_text:
|
|
filters.append(AdminUser.display_name.contains(display_text))
|
|
if status_text:
|
|
filters.append(AdminUser.status == status_text)
|
|
total = db.scalar(select(func.count()).select_from(AdminUser).where(*filters)) or 0
|
|
rows = db.scalars(
|
|
select(AdminUser).where(*filters).order_by(AdminUser.id.desc()).offset((current - 1) * size).limit(size)
|
|
).all()
|
|
role_map = _role_map(db, [row.id for row in rows])
|
|
return {
|
|
"records": [serialize_user(row, role_map.get(row.id, [])) for row in rows],
|
|
"current": current,
|
|
"size": size,
|
|
"total": total,
|
|
}
|
|
|
|
|
|
def create_user(
|
|
db: Session,
|
|
*,
|
|
username: str,
|
|
password: str,
|
|
display_name: str,
|
|
mobile: str | None,
|
|
role_ids: list[int],
|
|
) -> dict:
|
|
username_text = _require_text(username, empty="请填写用户名", limit=64, too_long="用户名过长")
|
|
display_text = _require_text(display_name, empty="请填写姓名", limit=64, too_long="姓名过长")
|
|
validate_password(password)
|
|
checked_roles = ensure_role_ids(db, role_ids)
|
|
if db.scalar(select(AdminUser.id).where(AdminUser.username == username_text)):
|
|
_bad("用户名已存在")
|
|
now = utcnow()
|
|
user = AdminUser(
|
|
username=username_text,
|
|
password_hash=hash_password(password),
|
|
display_name=display_text,
|
|
mobile_mask="",
|
|
status="active",
|
|
failed_login_count=0,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
if mobile:
|
|
apply_mobile(user, mobile)
|
|
db.add(user)
|
|
try:
|
|
db.flush()
|
|
except IntegrityError:
|
|
db.rollback()
|
|
_bad("用户名已存在")
|
|
replace_user_roles(db, user.id, checked_roles)
|
|
commit_or_raise(db, "用户名已存在")
|
|
roles = _role_map(db, [user.id]).get(user.id, [])
|
|
return serialize_user(user, roles)
|
|
|
|
|
|
def update_user(
|
|
db: Session,
|
|
user_id: int,
|
|
actor_id: int,
|
|
*,
|
|
display_name: str,
|
|
status_value: str,
|
|
role_ids: list[int],
|
|
mobile: str | None,
|
|
mobile_provided: bool,
|
|
) -> dict:
|
|
user = db.get(AdminUser, user_id)
|
|
if user is None:
|
|
_bad("用户不存在", status.HTTP_404_NOT_FOUND)
|
|
display_text = _require_text(display_name, empty="请填写姓名", limit=64, too_long="姓名过长")
|
|
if status_value not in USER_STATUSES:
|
|
_bad("用户状态不正确")
|
|
if status_value in {"disabled", "cancelled"} and user.id == actor_id:
|
|
_bad(KEEP_SELF)
|
|
checked_roles = ensure_role_ids(db, role_ids)
|
|
normalized_mobile = mobile
|
|
if mobile_provided and mobile is not None:
|
|
text = mobile.strip()
|
|
if len(text) > 20:
|
|
_bad("手机号过长")
|
|
normalized_mobile = text
|
|
|
|
def mutate() -> None:
|
|
user.display_name = display_text
|
|
user.status = status_value
|
|
if mobile_provided and normalized_mobile is not None:
|
|
apply_mobile(user, normalized_mobile)
|
|
replace_user_roles(db, user.id, checked_roles)
|
|
user.updated_at = utcnow()
|
|
|
|
preserve_full_admin(db, mutate)
|
|
commit_or_raise(db, "保存用户失败")
|
|
roles = _role_map(db, [user.id]).get(user.id, [])
|
|
return serialize_user(user, roles)
|
|
|
|
|
|
def reset_user_password(db: Session, user_id: int, password: str) -> dict:
|
|
user = db.get(AdminUser, user_id)
|
|
if user is None:
|
|
_bad("用户不存在", status.HTTP_404_NOT_FOUND)
|
|
validate_password(password)
|
|
user.password_hash = hash_password(password)
|
|
user.updated_at = utcnow()
|
|
commit_or_raise(db, "保存密码失败")
|
|
return {"ok": True}
|
|
|
|
|
|
def query_roles(
|
|
db: Session,
|
|
*,
|
|
current: int,
|
|
size: int,
|
|
name: str | None,
|
|
code: str | None,
|
|
status_value: str | None,
|
|
) -> dict:
|
|
current, size = _page(current, size)
|
|
filters = []
|
|
name_text = _exact(name)
|
|
code_text = _exact(code)
|
|
status_text = _exact(status_value)
|
|
if name_text:
|
|
filters.append(AdminRole.name.contains(name_text))
|
|
if code_text:
|
|
filters.append(AdminRole.code.contains(code_text))
|
|
if status_text:
|
|
filters.append(AdminRole.status == status_text)
|
|
total = db.scalar(select(func.count()).select_from(AdminRole).where(*filters)) or 0
|
|
rows = db.scalars(
|
|
select(AdminRole).where(*filters).order_by(AdminRole.id.desc()).offset((current - 1) * size).limit(size)
|
|
).all()
|
|
resource_map = _resource_id_map(db, [row.id for row in rows])
|
|
return {
|
|
"records": [serialize_role(row, resource_map.get(row.id, [])) for row in rows],
|
|
"current": current,
|
|
"size": size,
|
|
"total": total,
|
|
}
|
|
|
|
|
|
def create_role(db: Session, *, code: str, name: str, description: str | None) -> dict:
|
|
code_text = _require_text(code, empty="请填写角色编码", limit=64, too_long="角色编码过长")
|
|
name_text = _require_text(name, empty="请填写角色名称", limit=64, too_long="角色名称过长")
|
|
description_text = _optional_text(description, 2000, "说明过长")
|
|
if db.scalar(select(AdminRole.id).where(AdminRole.code == code_text)):
|
|
_bad("角色编码已存在")
|
|
now = utcnow()
|
|
role = AdminRole(
|
|
code=code_text,
|
|
name=name_text,
|
|
description=description_text,
|
|
status="active",
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
db.add(role)
|
|
commit_or_raise(db, "角色编码已存在")
|
|
return serialize_role(role, [])
|
|
|
|
|
|
def update_role(
|
|
db: Session,
|
|
role_id: int,
|
|
*,
|
|
name: str,
|
|
description: str | None,
|
|
status_value: str,
|
|
) -> dict:
|
|
role = db.get(AdminRole, role_id)
|
|
if role is None:
|
|
_bad("角色不存在", status.HTTP_404_NOT_FOUND)
|
|
name_text = _require_text(name, empty="请填写角色名称", limit=64, too_long="角色名称过长")
|
|
description_text = _optional_text(description, 2000, "说明过长")
|
|
if status_value not in ROLE_STATUSES:
|
|
_bad("角色状态不正确")
|
|
|
|
def mutate() -> None:
|
|
role.name = name_text
|
|
role.description = description_text
|
|
role.status = status_value
|
|
role.updated_at = utcnow()
|
|
|
|
preserve_full_admin(db, mutate)
|
|
commit_or_raise(db, "保存角色失败")
|
|
resource_ids = _resource_id_map(db, [role.id]).get(role.id, [])
|
|
return serialize_role(role, resource_ids)
|
|
|
|
|
|
def replace_role_resources(db: Session, role_id: int, resource_ids: list[int]) -> dict:
|
|
role = db.get(AdminRole, role_id)
|
|
if role is None:
|
|
_bad("角色不存在", status.HTTP_404_NOT_FOUND)
|
|
checked = ensure_resource_ids(db, resource_ids)
|
|
|
|
def mutate() -> None:
|
|
replace_role_links(db, role.id, checked)
|
|
role.updated_at = utcnow()
|
|
|
|
preserve_full_admin(db, mutate)
|
|
commit_or_raise(db, "保存角色资源失败")
|
|
return {"ok": True}
|
|
|
|
|
|
def list_resources(db: Session) -> list[dict]:
|
|
rows = list(db.scalars(select(AdminResource)).all())
|
|
return build_tree(rows)
|
|
|
|
|
|
def _resource_fields(
|
|
*,
|
|
resource_type: str,
|
|
name: str,
|
|
title: str,
|
|
path: str | None,
|
|
component: str | None,
|
|
icon: str | None,
|
|
permission: str | None,
|
|
sort: int,
|
|
visible: bool,
|
|
status_value: str,
|
|
) -> dict:
|
|
if resource_type not in RESOURCE_TYPES:
|
|
_bad("资源类型不正确")
|
|
if status_value not in RESOURCE_STATUSES:
|
|
_bad("资源状态不正确")
|
|
return {
|
|
"type": resource_type,
|
|
"name": _require_text(name, empty="请填写名称", limit=64, too_long="名称过长"),
|
|
"title": _require_text(title, empty="请填写标题", limit=64, too_long="标题过长"),
|
|
"path": _clean(path, 255, "路由过长"),
|
|
"component": _clean(component, 255, "组件路径过长"),
|
|
"icon": _clean(icon, 64, "图标过长"),
|
|
"permission": _clean(permission, 64, "权限码过长"),
|
|
"sort": sort,
|
|
"visible": bool(visible),
|
|
"status": status_value,
|
|
}
|
|
|
|
|
|
def create_resource(db: Session, payload: dict) -> dict:
|
|
parent_id = payload["parent_id"]
|
|
fields = _resource_fields(
|
|
resource_type=payload["type"],
|
|
name=payload["name"],
|
|
title=payload["title"],
|
|
path=payload.get("path"),
|
|
component=payload.get("component"),
|
|
icon=payload.get("icon"),
|
|
permission=payload.get("permission"),
|
|
sort=payload.get("sort") or 0,
|
|
visible=payload.get("visible", True),
|
|
status_value=payload.get("status") or "active",
|
|
)
|
|
assert_parent(db, None, parent_id)
|
|
if sibling_name_taken(db, parent_id, fields["name"]):
|
|
_bad("同级名称已存在")
|
|
resource = AdminResource(parent_id=parent_id, **fields)
|
|
db.add(resource)
|
|
commit_or_raise(db, "同级名称已存在")
|
|
return serialize_node(resource, [])
|
|
|
|
|
|
def update_resource(db: Session, resource_id: int, payload: dict) -> dict:
|
|
resource = db.get(AdminResource, resource_id)
|
|
if resource is None:
|
|
_bad("资源不存在", status.HTTP_404_NOT_FOUND)
|
|
parent_id = payload["parent_id"]
|
|
fields = _resource_fields(
|
|
resource_type=payload["type"],
|
|
name=payload["name"],
|
|
title=payload["title"],
|
|
path=payload.get("path"),
|
|
component=payload.get("component"),
|
|
icon=payload.get("icon"),
|
|
permission=payload.get("permission"),
|
|
sort=payload.get("sort") if payload.get("sort") is not None else resource.sort,
|
|
visible=payload.get("visible", resource.visible),
|
|
status_value=payload.get("status") or resource.status,
|
|
)
|
|
assert_parent(db, resource.id, parent_id)
|
|
if sibling_name_taken(db, parent_id, fields["name"], exclude_id=resource.id):
|
|
_bad("同级名称已存在")
|
|
resource.parent_id = parent_id
|
|
for key, value in fields.items():
|
|
setattr(resource, key, value)
|
|
commit_or_raise(db, "同级名称已存在")
|
|
return serialize_node(resource, [])
|
|
|
|
|
|
def delete_resource(db: Session, resource_id: int) -> dict:
|
|
resource = db.get(AdminResource, resource_id)
|
|
if resource is None:
|
|
_bad("资源不存在", status.HTTP_404_NOT_FOUND)
|
|
child_id = db.scalar(select(AdminResource.id).where(AdminResource.parent_id == resource_id).limit(1))
|
|
if child_id is not None:
|
|
_bad("存在子节点，不能删除")
|
|
db.execute(delete(AdminRoleResource).where(AdminRoleResource.resource_id == resource_id))
|
|
db.delete(resource)
|
|
commit_or_raise(db, "存在子节点，不能删除")
|
|
return {"ok": True}
|