基于FastAPI和Vue3的Python管理后台设计源码

import json import time from abc import ABC, abstractmethod from typing import Union from fastapi import Depends, Request from fastapi_pagination.bases import AbstractPage from fastapi_pagination.ext.databases import paginate from sqlalchemy import select from like.admin.config import AdminConfig from like.admin.schemas.system import ( SystemAuthAdminCreateIn, SystemAuthAdminEditIn, SystemAuthAdminUpdateIn, SystemAuthAdminListIn, SystemAuthAdminOut, SystemAuthAdminSelfOneOut, SystemAuthAdminSelfOut, SystemAuthAdminDetailOut) from like.dependencies.database import db from like.exceptions.base import AppException from like.http_base import HttpResp from like.models import ( system_auth_admin, system_auth_menu, system_auth_role, system_auth_dept, SystemAuthAdmin) from like.utils.redis import RedisUtil from like.utils.tools import ToolsUtil from like.utils.urls import UrlUtil from .auth_perm import ISystemAuthPermService, SystemAuthPermService from .auth_role import ISystemAuthRoleService, SystemAuthRoleService class ISystemAuthAdminService(ABC): """系统管理员服务抽象类""" @abstractmethod async def find_by_username(self, username: str) -> Union[SystemAuthAdmin, None]: pass @abstractmethod async def self(self, admin_id: int) -> SystemAuthAdminSelfOut: pass @abstractmethod async def list(self, list_in: SystemAuthAdminListIn) -> AbstractPage[SystemAuthAdminOut]: pass @abstractmethod async def detail(self, id_: int) -> SystemAuthAdminOut: pass @abstractmethod async def add(self, admin_create_in: SystemAuthAdminCreateIn): pass @abstractmethod async def edit(self, admin_edit_in: SystemAuthAdminEditIn): pass @abstractmethod async def update(self, admin_update_in: SystemAuthAdminUpdateIn, admin_id: int): pass @abstractmethod async def delete(self, id_: int): pass @abstractmethod async def disable(self, id_: int): pass @classmethod @abstractmethod async def cache_admin_user_by_uid(cls, id_: int): pass class SystemAuthAdminService(ISystemAuthAdminService): """系统管理员服务实现类""" async def find_by_username(self, username: str) -> Union[SystemAuthAdmin, None]: """根据账号查找管理员""" row = await db.fetch_one( system_auth_admin.select().where(system_auth_admin.c.username == username).limit(1)) return SystemAuthAdmin(**row) if row else None async def self(self, admin_id: int) -> SystemAuthAdminSelfOut: """当前管理员""" # 管理员信息 sys_admin = await db.fetch_one( system_auth_admin.select().where( system_auth_admin.c.id == admin_id, system_auth_admin.c.is_delete == 0).limit(1)) # 角色权限 auths = [] if admin_id > 1: menu_ids = await self.auth_perm_service.select_menu_ids_by_role_id( [int(i) for i in sys_admin.role_ids.split(',')]) if menu_ids: menus = await db.fetch_all( system_auth_menu.select() .where(system_auth_menu.c.id.in_(menu_ids), system_auth_menu.c.is_disable == 0, system_auth_menu.c.menu_type.in_(['C', 'A'])) .order_by(system_auth_menu.c.menu_sort, system_auth_menu.c.id)) if menus: auths.extend((i.perms.strip() for i in menus if i)) if not auths: auths.append('') else: auths.append('*') admin = SystemAuthAdminSelfOneOut.from_orm(sys_admin) admin.avatar = await UrlUtil.to_absolute_url(admin.avatar) return SystemAuthAdminSelfOut(user=admin, permissions=auths) async def list(self, list_in: SystemAuthAdminListIn) -> AbstractPage[SystemAuthAdminOut]: """管理员列表""" columns = [system_auth_admin.c.id, system_auth_admin.c.dept_ids.label('dept'), system_auth_admin.c.role_ids.label('role'), system_auth_admin.c.username, system_auth_admin.c.nickname, system_auth_admin.c.avatar, system_auth_admin.c.is_multipoint, system_auth_admin.c.is_disable, system_auth_admin.c.last_login_ip, system_auth_admin.c.last_login_time, system_auth_admin.c.create_time, system_auth_admin.c.update_time] # 查询条件 where = [system_auth_admin.c.is_delete == 0] if list_in.username: where.append(system_auth_admin.c.username.like(f'%{list_in.username}%')) if list_in.nickname: where.append(system_auth_admin.c.nickname.like(f'%{list_in.nickname}%')) if list_in.role is not None: where.append(system_auth_admin.c.role.in_(list_in.role)) query = select(columns).where(*where) \ .select_from(system_auth_admin) \ .order_by(system_auth_admin.c.id.desc(), system_auth_admin.c.sort.desc()) pager = await paginate(db, query) # 处理返回结果 for obj in pager.lists: obj.avatar = await UrlUtil.to_absolute_url(obj.avatar) if obj.id == 1: obj.role = '系统管理员' else: role_ids = [int(i) for i in obj.role.split(',') if i.isdigit()] roles = await db.fetch_all(system_auth_role.select().where(system_auth_role.c.id.in_(role_ids))) obj.role = '/'.join([i.name for i in roles]) if not obj.dept: obj.dept = '' else: dept_ids = [int(i) for i in obj.dept.split(',') if i.isdigit()] depts = await db.fetch_all(system_auth_dept.select().where( system_auth_dept.c.id.in_(dept_ids), system_auth_dept.c.is_delete == 0)) obj.dept = '/'.join([i.name for i in depts]) return pager async def detail(self, id_: int) -> SystemAuthAdminDetailOut: """管理员详细""" sys_admin = await db.fetch_one( system_auth_admin.select().where( system_auth_admin.c.id == id_, system_auth_admin.c.is_delete == 0).limit(1)) assert sys_admin, '账号已不存在！' sys_admin_out = SystemAuthAdminDetailOut.from_orm(sys_admin) sys_admin_out.avatar = await UrlUtil.to_absolute_url(sys_admin_out.avatar) sys_admin_out.roleIds = [int(i) for i in sys_admin_out.roleIds.split(',') if i.isdigit()] sys_admin_out.deptIds = [int(i) for i in sys_admin_out.deptIds.split(',') if i.isdigit()] sys_admin_out.postIds = [int(i) for i in sys_admin_out.postIds.split(',') if i.isdigit()] return sys_admin_out async def add(self, admin_create_in: SystemAuthAdminCreateIn): """管理员新增""" assert not await db.fetch_one( system_auth_admin.select() .where(system_auth_admin.c.username == admin_create_in.username, system_auth_admin.c.is_delete == 0).limit(1)), '账号已存在换一个吧！' assert not await db.fetch_one( system_auth_admin.select() .where(system_auth_admin.c.nickname == admin_create_in.nickname, system_auth_admin.c.is_delete == 0).limit(1)), '昵称已存在换一个吧！' if not (6 <= len(admin_create_in.password) <= 20): raise AppException(HttpResp.FAILED, msg='密码必须在6~20位') create_dict = dict(admin_create_in) salt = ToolsUtil.random_string(5) create_dict['role_ids'] = ','.join([str(i) for i in admin_create_in.role_ids]) create_dict['dept_ids'] = ','.join([str(i) for i in admin_create_in.dept_ids]) create_dict['post_ids'] = ','.join([str(i) for i in admin_create_in.post_ids]) create_dict['salt'] = salt create_dict['password'] = ToolsUtil.make_md5(f'{admin_create_in.password.strip()}