public/FastapiAdmin
1
0
Fork 0
mirror of https://github.com/fastapiadmin/FastapiAdmin.git synced 2026-09-21 12:52:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore: 清理冗余代码与配置，优化项目结构

Browse Source 
1.  删除无用文件与废弃代码：移除locale枚举、element-plus插件、sse路由、api token模块等
2.  简化类型导入与依赖：移除大量未使用的类型导入，统一echarts导入方式
3.  优化配置与样式：调整gitignore、样式引入顺序，新增列表动画样式
4.  修复接口与模型：修正接口返回类型、查询参数配置，更新部门模型字段
5.  优化性能与体验：添加图片懒加载，优化加载逻辑与表格渲染
6.  调整环境配置：新增并更新开发/生产环境配置文件
This commit is contained in:
zhangtao
2026-07-23 20:49:52 +08:00
parent 096e2216d6
commit 0ce31936aa
360 changed files with 8072 additions and 22122 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/v1/module_system/__init__.py
-2
View File
@@ -1,6 +1,5 @@
from fastapi import APIRouter
from app.api.v1.module_system.api_token.controller import ApiTokenRouter
from app.api.v1.module_system.auth.controller import AuthRouter
from app.api.v1.module_system.dept.controller import DeptRouter
from app.api.v1.module_system.dict.controller import DictRouter
@@ -27,5 +26,4 @@ system_router.include_router(RoleRouter)
system_router.include_router(TicketRouter)
system_router.include_router(UserRouter)
system_router.include_router(VersionRouter)
system_router.include_router(ApiTokenRouter)
system_router.include_router(MenuRouter)
backend/app/api/v1/module_system/api_token/__init__.py
-5
View File
@@ -1,5 +0,0 @@
"""API Token 模块：租户级访问令牌（用于外部系统/API 集成调用）。
模块路径：``module_system.api_token``，通过 module_system/__init__.py 注册路由。
外部 API 端点通过 ``api_external_router`` 注册到 ``/external`` 路径下。
"""
backend/app/api/v1/module_system/api_token/controller.py
-107
View File
@@ -1,107 +0,0 @@
"""API Token Controller：CRUD + reveal 二次验证
"""
from typing import Annotated
from fastapi import APIRouter, Body, Depends, Path, Query, Security
from fastapi.responses import JSONResponse
from sqlalchemy.ext.asyncio import AsyncSession
from app.common.response import ResponseSchema, SuccessResponse
from app.core.base_schema import AuthSchema, PageResultSchema, PaginationQueryParam
from app.core.dependencies import AuthPermission, db_getter
from app.core.router_class import OperationLogRoute
from .schema import (
ApiTokenCreatedSchema,
ApiTokenCreateSchema,
ApiTokenOutSchema,
ApiTokenQueryParam,
ApiTokenResetSchema,
ApiTokenRevealOutSchema,
ApiTokenRevealSchema,
)
from .service import ApiTokenService
ApiTokenRouter = APIRouter(route_class=OperationLogRoute, prefix="/api_token", tags=["系统-API令牌"])
@ApiTokenRouter.post("/create", summary="创建 API Token", response_model=ResponseSchema[ApiTokenCreatedSchema])
async def create_token_controller(
auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:api_token:create"]))],
db: Annotated[AsyncSession, Depends(db_getter)],
data: Annotated[ApiTokenCreateSchema, Body(description="创建参数")],
) -> JSONResponse:
"""创建后会完整返回明文 token，请立即保存。"""
result = await ApiTokenService(auth, db).create(data=data)
return SuccessResponse(data=result, msg="创建 token 成功")
@ApiTokenRouter.get("/list", summary="查询 token 列表", response_model=ResponseSchema[PageResultSchema[ApiTokenOutSchema]])
async def get_token_list_controller(
auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:api_token:query"]))],
db: Annotated[AsyncSession, Depends(db_getter)],
page: Annotated[PaginationQueryParam, Depends()],
search: Annotated[ApiTokenQueryParam, Query()],
) -> JSONResponse:
result = await ApiTokenService(auth, db).page(
page_no=page.page_no,
page_size=page.page_size,
search=search,
order_by=page.order_by,
)
return SuccessResponse(data=result, msg="查询成功")
@ApiTokenRouter.get("/detail/{id}", summary="token 详情", response_model=ResponseSchema[ApiTokenOutSchema])
async def get_token_detail_controller(
auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:api_token:detail"]))],
db: Annotated[AsyncSession, Depends(db_getter)],
id: Annotated[int, Path(description="token ID", ge=1)],
) -> JSONResponse:
result = await ApiTokenService(auth, db).detail(id=id)
return SuccessResponse(data=result, msg="查询成功")
@ApiTokenRouter.post("/{id}/reset", summary="重置 token（重新生成 secret）", response_model=ResponseSchema[ApiTokenCreatedSchema])
async def reset_token_controller(
auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:api_token:reset"]))],
db: Annotated[AsyncSession, Depends(db_getter)],
id: Annotated[int, Path(description="token ID", ge=1)],
data: Annotated[ApiTokenResetSchema, Body(description="可选项")],
) -> JSONResponse:
"""重置后会再次返回完整明文（仅此一次）。"""
result = await ApiTokenService(auth, db).reset(id=id, data=data)
return SuccessResponse(data=result, msg="重置 token 成功")
@ApiTokenRouter.patch("/{id}/status", summary="启用/禁用 token", response_model=ResponseSchema[None])
async def set_token_status_controller(
auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:api_token:patch"]))],
db: Annotated[AsyncSession, Depends(db_getter)],
id: Annotated[int, Path(description="token ID", ge=1)],
status: Annotated[int, Body(description="状态", ge=0, le=2)],
) -> JSONResponse:
await ApiTokenService(auth, db).set_status(id=id, status=status)
return SuccessResponse(msg="状态修改成功")
@ApiTokenRouter.delete("/{id}", summary="删除 token", response_model=ResponseSchema[None])
async def delete_token_controller(
auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:api_token:delete"]))],
db: Annotated[AsyncSession, Depends(db_getter)],
id: Annotated[int, Path(description="token ID", ge=1)],
) -> JSONResponse:
await ApiTokenService(auth, db).delete(id=id)
return SuccessResponse(msg="删除成功")
@ApiTokenRouter.post("/{id}/reveal", summary="查看 token 明文（需二次验证）", response_model=ResponseSchema[ApiTokenRevealOutSchema])
async def reveal_token_controller(
auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:api_token:reveal"]))],
db: Annotated[AsyncSession, Depends(db_getter)],
id: Annotated[int, Path(description="token ID", ge=1)],
data: Annotated[ApiTokenRevealSchema, Body(description="需输入当前用户密码")],
) -> JSONResponse:
"""高权限端点：会返回完整明文，需要二次密码验证。"""
result = await ApiTokenService(auth, db).reveal(id=id, data=data)
return SuccessResponse(data=result, msg="reveal 成功")
backend/app/api/v1/module_system/api_token/crud.py
-14
View File
@@ -1,14 +0,0 @@
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.base_crud import CRUDBase
from app.core.base_schema import AuthSchema
from .model import ApiTokenModel
from .schema import ApiTokenCreateSchema
class ApiTokenCRUD(CRUDBase[ApiTokenModel, ApiTokenCreateSchema, ApiTokenCreateSchema]):
"""平台 API Token CRUD 基础实现"""
def __init__(self, auth: AuthSchema, db: AsyncSession) -> None:
super().__init__(auth=auth, model=ApiTokenModel, db=db)
backend/app/api/v1/module_system/api_token/model.py
-40
View File
@@ -1,40 +0,0 @@
from datetime import datetime
from sqlalchemy import ForeignKey, Integer, String, Text
from sqlalchemy.orm import Mapped, mapped_column
from app.core.base_model import ModelMixin, UserMixin
class ApiTokenModel(ModelMixin, UserMixin):
"""API 访问令牌（用于外部系统/集成调用）
token 全名格式：
fastpat_<user_id_hex>_<48-char-base64url-secret>
字段：
- ``token_plain``：明文令牌（创建时一次性返回，后续可读但不推荐直接读）
- ``token_prefix``：用于列表展示的前 12 字符
- ``scopes``：JSON 数组（``["order:read", "user:write"]``），控制 API 可访问范围
- ``expires_at``：过期时间（空=永久）
- ``rate_limit``：每小时请求配额（默认 1000）
- ``status``：0=启用 1=禁用 2=吊销
- ``last_used_at/used_count/last_used_ip``：调用审计
"""
__tablename__: str = "sys_api_token"
__table_args__: dict[str, str] = {"comment": "API 访问令牌"}
__loader_options__: list[str] = ["created_by", "updated_by", "deleted_by"]
name: Mapped[str] = mapped_column(String(64), nullable=False, comment="令牌名称（业务语义，如：CRM-对账集成）")
token_prefix: Mapped[str] = mapped_column(String(32), nullable=False, index=True, comment="明文 token 前 12 字符（用于展示）")
token_plain: Mapped[str] = mapped_column(Text, nullable=False, comment="明文 token（自管理，按需用于外部集成）")
owner_user_id: Mapped[int | None] = mapped_column(Integer, ForeignKey("sys_user.id", ondelete="SET NULL", onupdate="CASCADE"), nullable=True, index=True, comment="所属用户ID（创建者/操作者）")
scopes: Mapped[str] = mapped_column(String(255), nullable=False, default="*", comment="可用 scope（逗号或 JSON 数组字符串）")
expires_at: Mapped[datetime | None] = mapped_column(nullable=True, comment="过期时间（NULL=永不过期）")
status: Mapped[int] = mapped_column(Integer, default=0, nullable=False, comment="状态(0:启用 1:禁用 2:吊销)", index=True)
rate_limit: Mapped[int] = mapped_column(Integer, default=1000, nullable=False, comment="每小时请求上限")
used_count: Mapped[int] = mapped_column(Integer, default=0, nullable=False, comment="累计调用次数")
last_used_at: Mapped[datetime | None] = mapped_column(nullable=True, comment="最近一次调用时间")
last_used_ip: Mapped[str | None] = mapped_column(String(64), nullable=True, comment="最近一次调用 IP")
description: Mapped[str | None] = mapped_column(Text, nullable=True, comment="备注")
backend/app/api/v1/module_system/api_token/schema.py
-89
View File
@@ -1,89 +0,0 @@
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field
from app.core.base_schema import PaginationQueryParam
class ApiTokenCreateSchema(BaseModel):
"""创建 API Token"""
name: str = Field(..., min_length=2, max_length=64, description="令牌业务名称")
scopes: list[str] = Field(default_factory=lambda: ["*"], description="可用 scope，``*`` 表示全部")
expires_at: datetime | None = Field(default=None, description="过期时间（NULL=永不过期）")
rate_limit: int = Field(default=1000, ge=1, le=1_000_000, description="每小时请求上限")
description: str | None = Field(default=None, max_length=512, description="备注")
class ApiTokenResetSchema(BaseModel):
"""重置（重新生成 secret）— 沿用同名 token，仅替换 secret 段"""
name: str | None = Field(default=None, max_length=64, description="新名称（不传则保持原值）")
scopes: list[str] | None = Field(default=None, description="新 scope（不传则保持原值）")
expires_at: datetime | None = Field(default=None, description="新过期时间")
rate_limit: int | None = Field(default=None, ge=1, le=1_000_000, description="新配额（NULL 保持原值）")
class ApiTokenQueryParam(PaginationQueryParam):
"""列表查询条件"""
name: str | None = Field(default=None, description="名称模糊匹配")
status: int | None = Field(default=None, description="状态精确匹配")
class ApiTokenRevealSchema(BaseModel):
"""查看明文 — 需当前用户密码二次验证"""
password: str = Field(..., min_length=6, max_length=128, description="当前用户登录密码")
class ApiTokenOutSchema(BaseModel):
"""列表/详情输出：脱敏（不含 token_plain）"""
model_config = ConfigDict(from_attributes=True)
id: int
name: str
token_prefix: str = Field(..., description="明文 token 前 12 字符，用于识别")
token_mask: str = Field(..., description="脱敏展示，例如 ``fastpat_xxxx****yz3w``")
owner_user_id: int | None
scopes: str
status: int
rate_limit: int
expires_at: datetime | None
used_count: int
last_used_at: datetime | None
last_used_ip: str | None
description: str | None
created_id: int | None
updated_id: int | None
created_time: datetime | None
updated_time: datetime | None
class ApiTokenCreatedSchema(BaseModel):
"""创建/重置响应：唯一含明文 token 的输出"""
model_config = ConfigDict(from_attributes=True)
id: int
name: str
token: str = Field(..., description="完整明文 token（仅此一次返回，请妥善保管）")
token_prefix: str
scopes: list[str]
expires_at: datetime | None
rate_limit: int
status: int
created_time: datetime | None
warning: str = Field(
default="请立即保存此 token。关闭此页面后将无法再次完整查看明文，如遗失请重置。",
description="安全提示",
)
class ApiTokenRevealOutSchema(BaseModel):
"""reveal 响应：含完整明文 + 警告"""
token: str
name: str
warning: str = "此为完整明文，仅高权限场景下返回，请勿写入日志/代码/对话。"
backend/app/api/v1/module_system/api_token/service.py
-298
View File
@@ -1,298 +0,0 @@
from __future__ import annotations
import json
import secrets
from datetime import datetime
from typing import Any
from redis.asyncio.client import Redis
from sqlalchemy import update as sa_update
from sqlalchemy.ext.asyncio import AsyncSession
from app.api.v1.module_system.user.crud import UserCRUD
from app.core.base_schema import AuthSchema, PageResultSchema
from app.core.database import async_db_session
from app.core.exceptions import CustomException
from app.core.logger import logger
from app.utils.password_util import PwdUtil
from .crud import ApiTokenCRUD
from .model import ApiTokenModel
from .schema import (
ApiTokenCreatedSchema,
ApiTokenCreateSchema,
ApiTokenOutSchema,
ApiTokenQueryParam,
ApiTokenResetSchema,
ApiTokenRevealOutSchema,
ApiTokenRevealSchema,
)
_TOKEN_PREFIX_HEADER = "fastpat_"
_TOKEN_PREFIX_DISPLAY_LEN = 12
_REDIS_RATE_KEY_PREFIX = "api_token:rate:"
def _generate_full_token(user_id: int) -> str:
"""生成完整 token：``fastpat_<user_id_hex>_<48-base64url>``"""
secret_part = secrets.token_urlsafe(36)
return f"{_TOKEN_PREFIX_HEADER}{user_id:x}_{secret_part}"
def _mask_token(full_token: str) -> str:
"""脱敏展示：保留头部 + ``****`` + 尾部 4 字符"""
if len(full_token) <= 16:
return "****"
return f"{full_token[:14]}****{full_token[-4:]}"
def _to_out_schema(token: ApiTokenModel) -> ApiTokenOutSchema:
return ApiTokenOutSchema(
id=token.id,
name=token.name,
token_prefix=token.token_prefix,
token_mask=_mask_token(token.token_plain),
owner_user_id=token.owner_user_id,
scopes=token.scopes,
status=token.status,
rate_limit=token.rate_limit,
expires_at=token.expires_at,
used_count=token.used_count,
last_used_at=token.last_used_at,
last_used_ip=token.last_used_ip,
description=token.description,
created_id=token.created_id,
updated_id=token.updated_id,
created_time=token.created_time,
updated_time=token.updated_time,
)
def _parse_scopes(scopes_str: str) -> list[str]:
if not scopes_str:
return []
if scopes_str == "*":
return ["*"]
try:
loaded = json.loads(scopes_str)
if isinstance(loaded, list):
return loaded
except (json.JSONDecodeError, ValueError):
pass
return [s.strip() for s in scopes_str.split(",") if s.strip()]
# ──────────────────────────────────────────────────────────
# Service
# ──────────────────────────────────────────────────────────
class ApiTokenService:
"""API Token 业务逻辑层"""
MAX_TOKENS_PER_USER: int = 50
def __init__(self, auth: AuthSchema, db: AsyncSession) -> None:
self.auth = auth
self.db = db
# ── 创建 ──────────────────────────────────────────────
async def create(self, data: ApiTokenCreateSchema) -> ApiTokenCreatedSchema:
existing = await ApiTokenCRUD(self.auth, self.db).get_list(
search={},
)
active_count = sum(1 for t in existing if t.status == 0 and not t.is_deleted)
if active_count >= self.MAX_TOKENS_PER_USER:
raise CustomException(msg=f"API Token 数量已达上限 ({self.MAX_TOKENS_PER_USER})，请先删除或禁用旧 token")
full_token = _generate_full_token(user_id=self.auth.user.id)
token_prefix = full_token[:_TOKEN_PREFIX_DISPLAY_LEN]
scopes_str = ",".join(data.scopes) if data.scopes else "*"
crud = ApiTokenCRUD(self.auth, self.db)
token_obj = await crud.create(
data={ # pyright: ignore[reportArgumentType]
"name": data.name,
"token_prefix": token_prefix,
"token_plain": full_token,
"owner_user_id": self.auth.user.id,
"scopes": scopes_str,
"expires_at": data.expires_at,
"status": 0,
"rate_limit": data.rate_limit,
"description": data.description,
},
)
if not token_obj:
raise CustomException(msg="创建 token 失败")
logger.info(f"新 token 创建成功: id={token_obj.id} name={data.name}")
return ApiTokenCreatedSchema(
id=token_obj.id,
name=token_obj.name,
token=full_token,
token_prefix=token_prefix,
scopes=data.scopes,
expires_at=token_obj.expires_at,
rate_limit=token_obj.rate_limit,
status=token_obj.status,
created_time=token_obj.created_time,
)
# ── 查询 ──────────────────────────────────────────────
async def page(
self,
page_no: int,
page_size: int,
search: ApiTokenQueryParam,
order_by: list[dict[str, str]] | None = None,
) -> PageResultSchema[ApiTokenOutSchema]:
crud = ApiTokenCRUD(self.auth, self.db)
search_dict: dict[str, Any] = {}
if search.name:
search_dict["name"] = ("like", f"%{search.name}%")
if search.status is not None:
search_dict["status"] = search.status
result = await crud.page(
offset=(page_no - 1) * page_size,
limit=page_size,
search=search_dict,
order_by=order_by or [{"id": "asc"}],
)
items_out = [_to_out_schema(row) for row in result.items]
return PageResultSchema[ApiTokenOutSchema](
page_no=result.page_no,
page_size=result.page_size,
total=result.total,
has_next=result.has_next,
items=items_out,
)
async def detail(self, id: int) -> ApiTokenOutSchema:
crud = ApiTokenCRUD(self.auth, self.db)
token = await crud.get_or_404(id=id)
return _to_out_schema(token)
# ── 状态/重置 ──────────────────────────────────────────
async def reset(self, id: int, data: ApiTokenResetSchema) -> ApiTokenCreatedSchema:
crud = ApiTokenCRUD(self.auth, self.db)
token = await crud.get_or_404(id=id)
full_token = _generate_full_token(user_id=self.auth.user.id)
token_prefix_new = full_token[:_TOKEN_PREFIX_DISPLAY_LEN]
values: dict[str, Any] = {
"token_prefix": token_prefix_new,
"token_plain": full_token,
"used_count": 0,
}
if data.name is not None:
values["name"] = data.name
if data.scopes is not None:
values["scopes"] = ",".join(data.scopes)
if data.expires_at is not None:
values["expires_at"] = data.expires_at
if data.rate_limit is not None:
values["rate_limit"] = data.rate_limit
await self.db.execute(sa_update(ApiTokenModel).where(ApiTokenModel.id == id).values(**values))
await self.db.flush()
await self.db.refresh(token)
logger.info(f"token[{id}] 已重置，新前缀={token_prefix_new}")
return ApiTokenCreatedSchema(
id=token.id,
name=token.name,
token=full_token,
token_prefix=token_prefix_new,
scopes=_parse_scopes(token.scopes),
expires_at=token.expires_at,
rate_limit=token.rate_limit,
status=token.status,
created_time=token.created_time,
)
async def set_status(self, id: int, status: int) -> None:
if status not in (0, 1, 2):
raise CustomException(msg="状态值不合法（0:启用 1:禁用 2:吊销）")
crud = ApiTokenCRUD(self.auth, self.db)
await crud.get_or_404(id=id)
await crud.update(id=id, data={"status": status}) # pyright: ignore[reportArgumentType]
async def delete(self, id: int) -> None:
crud = ApiTokenCRUD(self.auth, self.db)
await crud.get_or_404(id=id)
await crud.delete(ids=[id])
# ── reveal：二次验证后展示明文 ─────────────────────────
async def reveal(self, id: int, data: ApiTokenRevealSchema) -> ApiTokenRevealOutSchema:
user_row = await UserCRUD(self.auth, self.db).get(id=self.auth.user.id)
if not user_row:
raise CustomException(msg="用户不存在")
if not PwdUtil.verify_password(plain_password=data.password, password_hash=user_row.password):
logger.warning(f"reveal 二次验证失败: user_id={self.auth.user.id}")
raise CustomException(msg="密码错误，无法 reveal 明文")
crud = ApiTokenCRUD(self.auth, self.db)
token = await crud.get_or_404(id=id)
return ApiTokenRevealOutSchema(token=token.token_plain, name=token.name)
# ──────────────────────────────────────────────────────────
# 外部 API Bearer 验证（公开接口）
# ──────────────────────────────────────────────────────────
async def authenticate_api_token(token: str, request_ip: str | None = None, redis: Redis | None = None) -> ApiTokenModel:
"""外部 API 鉴权：从 Authorization Bearer 中解析 fastpat token，记录调用次数。"""
if not token or not token.startswith(_TOKEN_PREFIX_HEADER):
raise CustomException(msg="API Token 格式不合法", code=10401, status_code=401)
async with async_db_session() as db:
crud = ApiTokenCRUD(AuthSchema(), db)
candidate = await crud.get_list(search={"token_plain": ("=", token)})
if not candidate:
raise CustomException(msg="API Token 无效", code=10401, status_code=401)
token_row = candidate[0]
if token_row.status != 0:
raise CustomException(msg="API Token 已禁用或吊销", code=10401, status_code=401)
if token_row.expires_at is not None and token_row.expires_at < datetime.now():
raise CustomException(msg="API Token 已过期", code=10401, status_code=401)
if token_row.is_deleted:
raise CustomException(msg="API Token 已删除", code=10401, status_code=401)
# 限流（每小时）
if redis is not None:
try:
key = f"{_REDIS_RATE_KEY_PREFIX}{token_row.id}:{datetime.now().strftime('%Y%m%d%H')}"
current = await redis.incr(key)
if current == 1:
await redis.expire(key, 3600)
if current > token_row.rate_limit:
raise CustomException(
msg=f"API Token 限流：本小时已调用 {current} 次，上限 {token_row.rate_limit}",
code=10429,
status_code=429,
)
except CustomException:
raise
except Exception as e:
logger.warning(f"API Token 限流检查失败（继续放行）: {e!s}")
await db.execute(
sa_update(ApiTokenModel)
.where(ApiTokenModel.id == token_row.id)
.values(
used_count=ApiTokenModel.used_count + 1,
last_used_at=datetime.now(),
last_used_ip=request_ip,
),
)
await db.commit()
return token_row
backend/app/api/v1/module_system/auth/controller.py
+19
View File
@@ -1,3 +1,22 @@
"""
认证控制器 — TODO: 限流粒度细化
---------------------------------
当前登录（/login）和 OAuth 端点（/oauth/*）共享应用的通用限流配置，
缺少独立的、更严格的限流策略。建议为以下端点配置独立的 RateLimiter：
1. /auth/login — 密码登录
- 建议: 按 IP + 用户名组合限流，如 5次/分钟/IP + 10次/15分钟/用户
- 原因: 暴力破解防护
2. /auth/oauth/* — 第三方 OAuth 登录/回调
- 建议: 按 IP 限流，如 10次/分钟/IP
- 原因: OAuth 流程可能触发多次重定向，频率稍高于登录
3. /auth/captcha/* — 验证码获取/校验
- 建议: 按 IP 限流，如 3次/分钟/IP
- 原因: 防止验证码遍历
"""
import json
import secrets
from typing import Annotated
backend/app/api/v1/module_system/auth/oauth_service.py
+6
View File
@@ -36,6 +36,12 @@ STATE_PREFIX = "oauth_state:"
def _callback_url(request: Request, provider: OAuthProvider) -> str:
root = str(request.base_url).rstrip("/")
# 域名白名单校验：防止 Host 头注入攻击重定向到恶意域名
allowed_hosts = settings.OAUTH_ALLOWED_HOSTS
if allowed_hosts and allowed_hosts != ["*"]:
host = request.url.hostname
if host is None or not any(host == allowed_host or host.endswith("." + allowed_host) for allowed_host in allowed_hosts):
raise CustomException(msg="非法的 OAuth 回调域名")
return f"{root}/system/auth/oauth/{provider}/callback"
backend/app/api/v1/module_system/auth/service.py
+1 -1
View File
@@ -146,7 +146,7 @@ class LoginService:
)
auth = AuthSchema()
user = await UserCRUD(auth, db).get(username=login_form.username)
user = await UserCRUD(auth, db).get(username=login_form.username, preload=["roles", "roles.menus"])
if not user:
await _write_login_log(
backend/app/api/v1/module_system/dept/controller.py
+2 -11
View File
@@ -1,9 +1,7 @@
from typing import Annotated
from fastapi import APIRouter, Body, Depends, Path, Query, Security, status
from fastapi import APIRouter, Body, Depends, Path, Security, status
from fastapi.responses import JSONResponse
from fastapi_cache import FastAPICache
from fastapi_cache.decorator import cache
from sqlalchemy.ext.asyncio import AsyncSession
from app.common.response import ResponseSchema, SuccessResponse
@@ -16,15 +14,12 @@ from .service import DeptService
DeptRouter = APIRouter(route_class=OperationLogRoute, prefix="/dept", tags=["部门管理"])
_DEPT_NS = "dept"
@DeptRouter.get("/tree", summary="查询部门树", response_model=ResponseSchema[list[DeptOutSchema]])
@cache(expire=300, namespace=_DEPT_NS)
async def get_dept_tree_controller(
auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:dept:query"]))],
db: Annotated[AsyncSession, Depends(db_getter)],
search: Annotated[DeptQueryParam, Query()],
search: Annotated[DeptQueryParam, Depends()],
) -> JSONResponse:
order_by = [{"order": "asc"}]
result_dict_tree = await DeptService(auth, db).tree(search=search, order_by=order_by)
@@ -48,7 +43,6 @@ async def create_obj_controller(
data: Annotated[DeptCreateSchema, Body(description="部门创建参数")],
) -> JSONResponse:
result_dict = await DeptService(auth, db).create(data=data)
await FastAPICache.clear(namespace=_DEPT_NS)
return SuccessResponse(data=result_dict, msg="创建部门成功")
@@ -60,7 +54,6 @@ async def update_obj_controller(
data: Annotated[DeptUpdateSchema, Body(description="部门修改参数")],
) -> JSONResponse:
result_dict = await DeptService(auth, db).update(id=id, data=data)
await FastAPICache.clear(namespace=_DEPT_NS)
return SuccessResponse(data=result_dict, msg="修改部门成功")
@@ -71,7 +64,6 @@ async def delete_obj_controller(
ids: Annotated[list[int], Body(description="ID列表")],
) -> JSONResponse:
await DeptService(auth, db).delete(ids=ids)
await FastAPICache.clear(namespace=_DEPT_NS)
return SuccessResponse(msg="删除部门成功")
@@ -82,5 +74,4 @@ async def batch_set_available_obj_controller(
data: Annotated[BatchSetAvailable, Body(description="状态设置")],
) -> JSONResponse:
await DeptService(auth, db).batch_set_available(data=data)
await FastAPICache.clear(namespace=_DEPT_NS)
return SuccessResponse(msg="批量修改部门状态成功")
backend/app/api/v1/module_system/dept/model.py
+7 -17
View File
@@ -11,22 +11,16 @@ if TYPE_CHECKING:
class DeptModel(ModelMixin, UserMixin):
"""部门模型
"""
"""部门模型"""
__tablename__: str = "sys_dept"
__table_args__: dict[str, str] = {"comment": "部门表"}
__tree_children_attr__: str = "children"
__loader_options__: list[str] = ["children", "created_by", "updated_by", "deleted_by"]
name: Mapped[str] = mapped_column(String(64), nullable=False, comment="部门名称")
status: Mapped[int] = mapped_column(Integer, default=0, nullable=False, comment="状态(0:启动 1:停用)", index=True)
name: Mapped[str] = mapped_column(String(64), nullable=False, index=True, comment="部门名称")
status: Mapped[int] = mapped_column(Integer, default=0, nullable=False, comment="状态(0:启动 1:停用)")
description: Mapped[str | None] = mapped_column(Text, default=None, nullable=True, comment="备注")
order: Mapped[int] = mapped_column(Integer, nullable=False, default=999, comment="显示排序")
order: Mapped[int] = mapped_column(Integer, nullable=False, default=999, index=True, comment="显示排序")
code: Mapped[str] = mapped_column(String(64), unique=True, nullable=False, comment="部门编码")
leader: Mapped[str | None] = mapped_column(String(32), default=None, comment="部门负责人")
phone: Mapped[str | None] = mapped_column(String(20), default=None, comment="手机")
email: Mapped[str | None] = mapped_column(String(128), default=None, comment="邮箱")
parent_id: Mapped[int | None] = mapped_column(
Integer,
ForeignKey("sys_dept.id", ondelete="SET NULL", onupdate="CASCADE"),
@@ -40,10 +34,6 @@ class DeptModel(ModelMixin, UserMixin):
foreign_keys=[parent_id],
uselist=False,
)
children: Mapped[list["DeptModel"]] = relationship(back_populates="parent", foreign_keys=[parent_id], lazy="selectin")
roles: Mapped[list["RoleModel"]] = relationship(secondary="sys_role_depts", back_populates="depts", lazy="selectin")
users: Mapped[list["UserModel"]] = relationship(
back_populates="dept",
foreign_keys="UserModel.dept_id",
lazy="selectin",
)
children: Mapped[list["DeptModel"]] = relationship(back_populates="parent", foreign_keys=[parent_id])
roles: Mapped[list["RoleModel"]] = relationship(secondary="sys_role_depts", back_populates="depts")
users: Mapped[list["UserModel"]] = relationship(back_populates="dept", foreign_keys="UserModel.dept_id")
backend/app/api/v1/module_system/dept/schema.py
+3 -6
View File
@@ -10,10 +10,7 @@ class DeptCreateSchema(BaseModel):
name: str = Field(..., min_length=1, max_length=64, description="部门名称")
order: int = Field(default=1, ge=0, description="显示顺序")
code: str = Field(..., min_length=2, max_length=64, description="部门编码")
leader: str | None = Field(default=None, max_length=32, description="部门负责人")
phone: str | None = Field(default=None, max_length=20, description="联系电话")
email: str | None = Field(default=None, max_length=128, description="邮箱")
parent_id: int | None = Field(default=None, ge=0, description="父部门ID")
parent_id: int | None = Field(default=None, ge=1, description="父部门ID")
status: int = Field(default=0, ge=0, le=1, description="状态(0:启动 1:停用)")
description: str | None = Field(default=None, max_length=255, description="备注")
@@ -61,5 +58,5 @@ class DeptTreeOutSchema(DeptOutSchema):
class DeptQueryParam(BaseQueryParam, UserByQueryParam):
"""部门管理查询参数"""
name: str | None = Field(None, description="部门名称")
status: int | None = Field(None, ge=0, le=1, description="状态(0:启动 1:停用)")
name: str | None = Field(None, description="部门名称", json_schema_extra={"q": "like"})
status: int | None = Field(None, ge=0, le=1, description="状态(0:启动 1:停用)", json_schema_extra={"q": "eq"})
backend/app/api/v1/module_system/dept/service.py
+8 -13
View File
@@ -8,6 +8,7 @@ from app.utils.common_util import (
get_parent_id_map,
get_parent_recursion,
search_to_dict,
traversal_to_tree,
)
from .crud import DeptCRUD
@@ -15,7 +16,6 @@ from .schema import (
DeptCreateSchema,
DeptOutSchema,
DeptQueryParam,
DeptTreeOutSchema,
DeptUpdateSchema,
)
@@ -44,9 +44,9 @@ class DeptService:
search: DeptQueryParam | None = None,
order_by: list[dict] | None = None,
) -> list[dict]:
dept_list = await DeptCRUD(self.auth, self.db).tree_list(search=search_to_dict(search), order_by=order_by)
dept_dict_list = [DeptTreeOutSchema.model_validate(dept).model_dump() for dept in dept_list]
return [d for d in dept_dict_list if d.get("parent_id") is None]
dept_list = await DeptCRUD(self.auth, self.db).get_list(search=search_to_dict(search), order_by=order_by)
dept_dict_list = [DeptOutSchema.model_validate(dept).model_dump() for dept in dept_list]
return traversal_to_tree(dept_dict_list)
async def create(self, data: DeptCreateSchema) -> DeptOutSchema:
dept = await DeptCRUD(self.auth, self.db).get(name=data.name)
@@ -57,10 +57,10 @@ class DeptService:
raise CustomException(msg="创建失败，编码已存在")
dept = await DeptCRUD(self.auth, self.db).create(data=data)
return DeptOutSchema.model_validate(dept)
return await self.detail(id=dept.id)
async def update(self, id: int, data: DeptUpdateSchema) -> DeptOutSchema:
dept = await DeptCRUD(self.auth, self.db).get_or_404(id=id, msg="更新失败，该数据不存在")
await DeptCRUD(self.auth, self.db).get_or_404(id=id, msg="更新失败，该数据不存在")
exist_dept = await DeptCRUD(self.auth, self.db).get(name=data.name)
if exist_dept and exist_dept.id != id:
raise CustomException(msg="更新失败，名称已存在")
@@ -68,13 +68,8 @@ class DeptService:
if exist_code and exist_code.id != id:
raise CustomException(msg="更新失败，编码已存在")
dept = await DeptCRUD(self.auth, self.db).update(id=id, data=data)
dept_out = DeptOutSchema.model_validate(dept)
if dept_out.parent_id:
parent = await DeptCRUD(self.auth, self.db).get(id=dept_out.parent_id)
if parent:
dept_out.parent_name = parent.name
return dept_out
await DeptCRUD(self.auth, self.db).update(id=id, data=data)
return await self.detail(id=id)
async def delete(self, ids: list[int]) -> None:
if not ids:
backend/app/api/v1/module_system/dict/controller.py
+5 -51
View File
@@ -1,17 +1,14 @@
from typing import Annotated
from fastapi import APIRouter, Body, Depends, Path, Query, Security, status
from fastapi.responses import JSONResponse, StreamingResponse
from fastapi_cache import FastAPICache
from fastapi_cache.decorator import cache
from fastapi import APIRouter, Body, Depends, Path, Security, status
from fastapi.responses import JSONResponse
from redis.asyncio.client import Redis
from sqlalchemy.ext.asyncio import AsyncSession
from app.common.response import ResponseSchema, StreamResponse, SuccessResponse
from app.common.response import ResponseSchema, SuccessResponse
from app.core.base_schema import AuthSchema, BatchSetAvailable, PageResultSchema, PaginationQueryParam
from app.core.dependencies import AuthPermission, db_getter, redis_getter
from app.core.router_class import OperationLogRoute
from app.utils.common_util import bytes2file_response
from .schema import (
DictDataCreateSchema,
@@ -27,8 +24,6 @@ from .service import DictDataService, DictTypeService
DictRouter = APIRouter(route_class=OperationLogRoute, prefix="/dict", tags=["字典管理"])
_DICT_TYPE_NS = "dict_type"
@DictRouter.get("/type/detail/{id}", summary="获取字典类型详情", response_model=ResponseSchema[DictTypeOutSchema])
async def get_type_detail_controller(
@@ -44,7 +39,7 @@ async def get_type_detail_controller(
async def get_type_list_controller(
auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:dict_type:query"]))],
page: Annotated[PaginationQueryParam, Depends()],
search: Annotated[DictTypeQueryParam, Query()],
search: Annotated[DictTypeQueryParam, Depends()],
db: Annotated[AsyncSession, Depends(db_getter)],
) -> JSONResponse:
result_dict = await DictTypeService(auth, db).page(
@@ -57,7 +52,6 @@ async def get_type_list_controller(
@DictRouter.get("/type/optionselect", summary="获取全部字典类型", response_model=ResponseSchema[list[DictTypeOutSchema]])
@cache(expire=300, namespace=_DICT_TYPE_NS)
async def get_type_optionselect_controller(
auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:dict_type:query"]))],
db: Annotated[AsyncSession, Depends(db_getter)],
@@ -74,7 +68,6 @@ async def create_type_controller(
data: Annotated[DictTypeCreateSchema, Body(description="字典类型创建参数")],
) -> JSONResponse:
result_dict = await DictTypeService(auth, db).create(redis=redis, data=data)
await FastAPICache.clear(namespace=_DICT_TYPE_NS)
return SuccessResponse(data=result_dict, msg="创建字典类型成功")
@@ -87,7 +80,6 @@ async def update_type_controller(
data: Annotated[DictTypeUpdateSchema, Body(description="字典类型修改参数")],
) -> JSONResponse:
result_dict = await DictTypeService(auth, db).update(redis=redis, id=id, data=data)
await FastAPICache.clear(namespace=_DICT_TYPE_NS)
return SuccessResponse(data=result_dict, msg="修改字典类型成功")
@@ -99,7 +91,6 @@ async def delete_type_controller(
ids: Annotated[list[int], Body(description="字典类型ID列表")],
) -> JSONResponse:
await DictTypeService(auth, db).delete(redis=redis, ids=ids)
await FastAPICache.clear(namespace=_DICT_TYPE_NS)
return SuccessResponse(msg="删除字典类型成功")
@@ -110,28 +101,9 @@ async def batch_set_available_dict_type_controller(
db: Annotated[AsyncSession, Depends(db_getter)],
) -> JSONResponse:
await DictTypeService(auth, db).set_available(data=data)
await FastAPICache.clear(namespace=_DICT_TYPE_NS)
return SuccessResponse(msg="批量修改字典类型状态成功")
@DictRouter.post("/type/export", summary="导出字典类型")
async def export_type_list_controller(
auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:dict_type:export"]))],
search: Annotated[DictTypeQueryParam, Body()],
db: Annotated[AsyncSession, Depends(db_getter)],
) -> StreamingResponse:
# 获取全量数据并转为dict列表
result_dict_list = await DictTypeService(auth, db).get_list(search=search)
export_data = [item.model_dump() for item in result_dict_list]
export_result = DictTypeService.export(data_list=export_data)
return StreamResponse(
data=bytes2file_response(export_result),
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
headers={"Content-Disposition": "attachment; filename=dict_type.xlsx"},
)
@DictRouter.get("/data/detail/{id}", summary="获取字典数据详情", response_model=ResponseSchema[DictDataOutSchema])
async def get_data_detail_controller(
auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:dict_data:detail"]))],
@@ -146,7 +118,7 @@ async def get_data_detail_controller(
async def get_data_list_controller(
auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:dict_data:query"]))],
page: Annotated[PaginationQueryParam, Depends()],
search: Annotated[DictDataQueryParam, Query()],
search: Annotated[DictDataQueryParam, Depends()],
db: Annotated[AsyncSession, Depends(db_getter)],
) -> JSONResponse:
order_by = [{"order": "asc"}]
@@ -205,24 +177,6 @@ async def batch_set_available_dict_data_controller(
return SuccessResponse(msg="批量修改字典数据状态成功")
@DictRouter.post("/data/export", summary="导出字典数据")
async def export_data_list_controller(
auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:dict_data:export"]))],
page: Annotated[PaginationQueryParam, Depends()],
search: Annotated[DictDataQueryParam, Body()],
db: Annotated[AsyncSession, Depends(db_getter)],
) -> StreamingResponse:
result_dict_list = await DictDataService(auth, db).get_list(search=search, order_by=page.order_by)
export_data = [item.model_dump() for item in result_dict_list]
export_result = DictDataService.export(data_list=export_data)
return StreamResponse(
data=bytes2file_response(export_result),
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
headers={"Content-Disposition": "attachment; filename=dice_data.xlsx"},
)
@DictRouter.get("/data/info/{dict_type}", summary="根据字典类型获取数据", response_model=ResponseSchema[list[DictDataOutSchema]])
async def get_init_dict_data_controller(
redis: Annotated[Redis, Depends(redis_getter)],
backend/app/api/v1/module_system/dict/model.py
+8 -14
View File
@@ -1,35 +1,29 @@
from sqlalchemy import Boolean, ForeignKey, Integer, String, Text
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.core.base_model import ModelMixin, UserMixin
from app.core.base_model import ModelMixin
class DictTypeModel(ModelMixin, UserMixin):
class DictTypeModel(ModelMixin):
"""字典类型表"""
__tablename__: str = "sys_dict_type"
__table_args__: dict[str, str] = {"comment": "字典类型表"}
__loader_options__: list[str] = ["dict_data_list", "created_by", "updated_by", "deleted_by"]
dict_name: Mapped[str] = mapped_column(String(64), nullable=False, comment="字典名称")
dict_type: Mapped[str] = mapped_column(String(255), nullable=False, index=True, comment="字典类型")
status: Mapped[int] = mapped_column(Integer, default=0, nullable=False, comment="状态(0:启动 1:停用)", index=True)
dict_name: Mapped[str] = mapped_column(String(100), nullable=False, index=True, comment="字典名称")
dict_type: Mapped[str] = mapped_column(String(255), nullable=False, index=True, unique=True, comment="字典类型")
status: Mapped[int] = mapped_column(Integer, default=0, nullable=False, comment="状态(0:启动 1:停用)")
description: Mapped[str | None] = mapped_column(Text, default=None, nullable=True, comment="备注")
dict_data_list: Mapped[list["DictDataModel"]] = relationship(
"DictDataModel",
back_populates="dict_type_obj",
cascade="all, delete-orphan",
)
dict_data_list: Mapped[list["DictDataModel"]] = relationship("DictDataModel", back_populates="dict_type_obj")
class DictDataModel(ModelMixin, UserMixin):
class DictDataModel(ModelMixin):
"""字典数据表"""
__tablename__: str = "sys_dict_data"
__table_args__: dict[str, str] = {"comment": "字典数据表"}
__loader_options__: list[str] = ["dict_type_obj", "created_by", "updated_by", "deleted_by"]
status: Mapped[int] = mapped_column(Integer, default=0, nullable=False, comment="状态(0:启动 1:停用)", index=True)
status: Mapped[int] = mapped_column(Integer, default=0, nullable=False, index=True, comment="状态(0:启动 1:停用)")
description: Mapped[str | None] = mapped_column(Text, default=None, nullable=True, comment="备注")
dict_sort: Mapped[int] = mapped_column(Integer, nullable=False, default=0, comment="字典排序")
dict_label: Mapped[str] = mapped_column(String(255), nullable=False, comment="字典标签")
backend/app/api/v1/module_system/dict/schema.py
+11 -11
View File
@@ -8,7 +8,7 @@ from pydantic import (
model_validator,
)
from app.core.base_schema import BaseQueryParam, BaseSchema, UserByQueryParam, UserBySchema
from app.core.base_schema import BaseQueryParam, BaseSchema
class DictTypeCreateSchema(BaseModel):
@@ -71,18 +71,18 @@ class DictTypeUpdateSchema(DictTypeCreateSchema):
"""字典类型更新模型"""
class DictTypeOutSchema(DictTypeCreateSchema, BaseSchema, UserBySchema):
class DictTypeOutSchema(DictTypeCreateSchema, BaseSchema):
"""字典类型响应模型"""
model_config = ConfigDict(from_attributes=True)
class DictTypeQueryParam(BaseQueryParam, UserByQueryParam):
class DictTypeQueryParam(BaseQueryParam):
"""字典类型查询参数"""
dict_name: str | None = Field(default=None, description="字典名称", max_length=100)
dict_name: str | None = Field(default=None, description="字典名称", max_length=100, json_schema_extra={"q": "like"})
dict_type: str | None = Field(default=None, description="字典类型", max_length=100, json_schema_extra={"q": "eq"})
status: int | None = Field(default=None, ge=0, le=1, description="状态(0:启动 1:停用)")
status: int | None = Field(default=None, ge=0, le=1, description="状态(0:启动 1:停用)", json_schema_extra={"q": "eq"})
class DictDataCreateSchema(BaseModel):
@@ -138,16 +138,16 @@ class DictDataUpdateSchema(DictDataCreateSchema):
"""字典数据更新模型"""
class DictDataOutSchema(DictDataCreateSchema, BaseSchema, UserBySchema):
class DictDataOutSchema(DictDataCreateSchema, BaseSchema):
"""字典数据响应模型"""
model_config = ConfigDict(from_attributes=True)
class DictDataQueryParam(BaseQueryParam, UserByQueryParam):
class DictDataQueryParam(BaseQueryParam):
"""字典数据查询参数"""
dict_label: str | None = Field(default=None, description="字典标签", max_length=100)
dict_type: str | None = Field(default=None, description="字典类型", max_length=100, json_schema_extra={"q": "eq"})
dict_type_id: int | None = Field(default=None, description="字典类型ID")
status: int | None = Field(default=None, ge=0, le=1, description="状态(0:启动 1:停用)")
dict_label: str | None = Field(default=None, description="字典标签", max_length=255, json_schema_extra={"q": "like"})
dict_type: str | None = Field(default=None, description="字典类型", max_length=255, json_schema_extra={"q": "eq"})
dict_type_id: int | None = Field(default=None, description="字典类型ID", json_schema_extra={"q": "eq"})
status: int | None = Field(default=None, ge=0, le=1, description="状态(0:启动 1:停用)", json_schema_extra={"q": "eq"})
backend/app/api/v1/module_system/dict/service.py
-67
View File
@@ -13,7 +13,6 @@ from app.core.exceptions import CustomException
from app.core.logger import logger
from app.core.redis_crud import RedisCURD
from app.utils.common_util import search_to_dict
from app.utils.excel_util import ExcelUtil
from .crud import DictDataCRUD, DictTypeCRUD
from .schema import (
@@ -234,37 +233,6 @@ class DictTypeService:
"""
await DictTypeCRUD(self.auth, self.db).set(ids=data.ids, status=data.status)
@staticmethod
def export(data_list: list[dict]) -> bytes:
"""导出数据字典类型列表（无状态工具方法）
参数:
- data_list (list[dict]): 数据字典类型列表
返回:
- bytes: Excel文件字节流
"""
mapping_dict = {
"id": "编号",
"dict_name": "字典名称",
"dict_type": "字典类型",
"status": "状态",
"description": "备注",
"created_time": "创建时间",
"updated_time": "更新时间",
"created_id": "创建者ID",
"updated_id": "更新者ID",
}
# 复制数据并转换状态
data = data_list.copy()
for item in data:
# 处理状态
item["status"] = "启用" if item.get("status") == 0 else "停用"
return ExcelUtil.export_list2excel(list_data=data, mapping_dict=mapping_dict)
class DictDataService:
"""字典数据管理服务
@@ -553,38 +521,3 @@ class DictDataService:
- None
"""
await DictDataCRUD(self.auth, self.db).set(ids=data.ids, status=data.status)
@staticmethod
def export(data_list: list[dict]) -> bytes:
"""导出数据字典数据列表（无状态工具方法）
参数:
- data_list (list[dict]): 数据字典数据列表
返回:
- bytes: Excel文件字节流
"""
mapping_dict = {
"id": "编号",
"dict_type": "字典类型",
"dict_label": "字典标签",
"dict_value": "字典键值",
"dict_sort": "字典排序",
"status": "状态",
"description": "备注",
"created_time": "创建时间",
"updated_time": "更新时间",
"created_id": "创建者ID",
"updated_id": "更新者ID",
}
# 复制数据并转换状态
data = data_list.copy()
for item in data:
item["status"] = "启用" if item.get("status") == 0 else "停用"
if item.get("is_default") is True:
item["is_default"] = "是"
else:
item["is_default"] = "否"
return ExcelUtil.export_list2excel(list_data=data, mapping_dict=mapping_dict)
backend/app/api/v1/module_system/log/controller.py
+3 -3
View File
@@ -1,6 +1,6 @@
from typing import Annotated
from fastapi import APIRouter, Body, Depends, Path, Query, Security
from fastapi import APIRouter, Body, Depends, Path, Security
from fastapi.responses import JSONResponse, StreamingResponse
from sqlalchemy.ext.asyncio import AsyncSession
@@ -38,7 +38,7 @@ async def get_log_list_controller(
auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:login_log:query"]))],
db: Annotated[AsyncSession, Depends(db_getter)],
page: Annotated[PaginationQueryParam, Depends()],
search: Annotated[LoginLogQueryParam, Query()],
search: Annotated[LoginLogQueryParam, Depends()],
) -> JSONResponse:
result_dict = await LoginLogService(auth, db).page(
page_no=page.page_no,
@@ -76,7 +76,7 @@ async def get_operation_log_list_controller(
auth: Annotated[AuthSchema, Depends(get_current_user)],
db: Annotated[AsyncSession, Depends(db_getter)],
page: Annotated[PaginationQueryParam, Depends()],
search: Annotated[OperationLogQueryParam, Query()],
search: Annotated[OperationLogQueryParam, Depends()],
) -> JSONResponse:
result_dict = await OperationLogService(auth, db).page(
page_no=page.page_no,
backend/app/api/v1/module_system/log/model.py
+6 -7
View File
@@ -27,9 +27,9 @@ class LoginLogModel(ModelMixin):
__tablename__: str = "sys_login_log"
__table_args__: dict[str, str] = {"comment": "登录日志表"}
status: Mapped[int] = mapped_column(Integer, default=1, comment="登录状态(1成功 2失败)", index=True)
status: Mapped[int] = mapped_column(Integer, default=1, index=True, comment="登录状态(1成功 2失败)")
description: Mapped[str | None] = mapped_column(Text, default=None, nullable=True, comment="备注")
username: Mapped[str] = mapped_column(String(64), nullable=False, comment="用户名")
username: Mapped[str] = mapped_column(String(64), nullable=False, index=True, comment="用户名")
login_location: Mapped[str | None] = mapped_column(String(255), nullable=True, comment="登录位置")
login_ip: Mapped[str | None] = mapped_column(String(50), nullable=True, comment="登录IP地址")
request_os: Mapped[str | None] = mapped_column(String(64), nullable=True, comment="操作系统")
@@ -38,16 +38,15 @@ class LoginLogModel(ModelMixin):
class OperationLogModel(ModelMixin):
"""操作日志模型
"""
"""操作日志模型"""
__tablename__: str = "sys_operation_log"
__table_args__: dict[str, str] = {"comment": "操作日志表"}
username: Mapped[str] = mapped_column(String(64), nullable=False, comment="操作人用户名")
status: Mapped[int] = mapped_column(Integer, default=0, nullable=False, comment="操作状态(0:成功 1:失败)", index=True)
username: Mapped[str] = mapped_column(String(64), nullable=False, index=True, comment="操作人用户名")
status: Mapped[int] = mapped_column(Integer, default=0, nullable=False, comment="操作状态(0:成功 1:失败)")
description: Mapped[str | None] = mapped_column(Text, default=None, nullable=True, comment="备注")
request_path: Mapped[str] = mapped_column(String(255), comment="请求路径")
request_path: Mapped[str] = mapped_column(String(255), index=True, comment="请求路径")
request_method: Mapped[str] = mapped_column(String(10), comment="请求方式")
request_payload: Mapped[str | None] = mapped_column(get_log_text_column_type(), comment="请求体")
response_code: Mapped[int] = mapped_column(Integer, comment="响应状态码")
backend/app/api/v1/module_system/log/schema.py
+7 -16
View File
@@ -1,6 +1,5 @@
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
from pydantic import BaseModel, ConfigDict, Field, field_validator
from app.common.enums import QueueEnum
from app.core.base_schema import BaseQueryParam, BaseSchema
ALLOWED_REQUEST_METHODS = ["GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS", "HEAD"]
@@ -48,26 +47,18 @@ class LoginLogDetailOutSchema(LoginLogOutSchema):
class LoginLogQueryParam(BaseQueryParam):
"""登录日志查询参数"""
username: str | tuple[str, str] | None = Field(None, max_length=64, description="用户名")
status: int | tuple[str, int] | None = Field(None, description="登录状态(1:成功 2:失败)")
@model_validator(mode="after")
def validate_query_params(self) -> "LoginLogQueryParam":
if isinstance(self.username, str):
self.username = (QueueEnum.like.value, self.username)
if isinstance(self.status, int):
self.status = (QueueEnum.eq.value, self.status)
return self
username: str | None = Field(None, max_length=64, description="用户名", json_schema_extra={"q": "like"})
status: int | None = Field(None, description="登录状态(1:成功 2:失败)", json_schema_extra={"q": "eq"})
class OperationLogQueryParam(BaseQueryParam):
"""操作日志查询参数"""
request_path: str | None = Field(None, description="请求路径")
request_path: str | None = Field(None, description="请求路径", json_schema_extra={"q": "like"})
request_method: str | None = Field(None, description="请求方式", json_schema_extra={"q": "eq"})
username: str | None = Field(None, description="用户名")
status: int | None = Field(None, ge=0, le=1, description="状态(0:成功 1:失败)")
request_ip: str | None = Field(None, description="请求IP")
username: str | None = Field(None, description="用户名", json_schema_extra={"q": "like"})
status: int | None = Field(None, ge=0, le=1, description="状态(0:成功 1:失败)", json_schema_extra={"q": "eq"})
request_ip: str | None = Field(None, description="请求IP", json_schema_extra={"q": "eq"})
class OperationLogOutSchema(BaseSchema):
backend/app/api/v1/module_system/menu/controller.py
+2 -11
View File
@@ -1,9 +1,7 @@
from typing import Annotated
from fastapi import APIRouter, Body, Depends, Path, Query, Security, status
from fastapi import APIRouter, Body, Depends, Path, Security, status
from fastapi.responses import JSONResponse
from fastapi_cache import FastAPICache
from fastapi_cache.decorator import cache
from sqlalchemy.ext.asyncio import AsyncSession
from app.common.response import ResponseSchema, SuccessResponse
@@ -16,15 +14,12 @@ from .service import MenuService
MenuRouter = APIRouter(route_class=OperationLogRoute, prefix="/menu", tags=["菜单管理"])
_MENU_NS = "menu"
@MenuRouter.get("/tree", summary="查询菜单树", response_model=ResponseSchema[list[MenuOutSchema]])
@cache(expire=300, namespace=_MENU_NS)
async def get_menu_tree_controller(
auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:menu:query"]))],
db: Annotated[AsyncSession, Depends(db_getter)],
search: Annotated[MenuQueryParam, Query()],
search: Annotated[MenuQueryParam, Depends()],
) -> JSONResponse:
order_by = [{"order": "asc"}]
result_dict_tree = await MenuService(auth, db).tree(search=search, order_by=order_by)
@@ -48,7 +43,6 @@ async def create_obj_controller(
data: Annotated[MenuCreateSchema, Body(description="菜单创建参数")],
) -> JSONResponse:
result_dict = await MenuService(auth, db).create(data=data)
await FastAPICache.clear(namespace=_MENU_NS)
return SuccessResponse(data=result_dict, msg="创建菜单成功")
@@ -60,7 +54,6 @@ async def update_obj_controller(
data: Annotated[MenuUpdateSchema, Body(description="菜单修改参数")],
) -> JSONResponse:
result_dict = await MenuService(auth, db).update(id=id, data=data)
await FastAPICache.clear(namespace=_MENU_NS)
return SuccessResponse(data=result_dict, msg="修改菜单成功")
@@ -71,7 +64,6 @@ async def delete_obj_controller(
ids: Annotated[list[int], Body(description="菜单ID列表")],
) -> JSONResponse:
await MenuService(auth, db).delete(ids=ids)
await FastAPICache.clear(namespace=_MENU_NS)
return SuccessResponse(msg="删除菜单成功")
@@ -82,5 +74,4 @@ async def batch_set_available_obj_controller(
data: Annotated[BatchSetAvailable, Body(description="状态设置")],
) -> JSONResponse:
await MenuService(auth, db).set_available(data=data)
await FastAPICache.clear(namespace=_MENU_NS)
return SuccessResponse(msg="批量修改菜单状态成功")
backend/app/api/v1/module_system/menu/model.py
+6 -8
View File
@@ -21,16 +21,14 @@ class MenuModel(ModelMixin):
__tablename__: str = "sys_menu"
__table_args__: dict[str, str] = {"comment": "系统菜单表"}
__tree_children_attr__: str = "children"
__loader_options__: list[str] = ["roles", "children"]
name: Mapped[str] = mapped_column(String(50), nullable=False, comment="菜单名称")
name: Mapped[str] = mapped_column(String(64), nullable=False, index=True, comment="菜单名称")
type: Mapped[int] = mapped_column(Integer, nullable=False, default=2, comment="菜单类型(1:目录 2:菜单 3:按钮 4:链接)")
order: Mapped[int] = mapped_column(Integer, nullable=False, default=999, comment="显示排序")
order: Mapped[int] = mapped_column(Integer, nullable=False, default=999, index=True, comment="显示排序")
permission: Mapped[str | None] = mapped_column(String(100), comment="权限标识(如:module_system:user:query)")
icon: Mapped[str | None] = mapped_column(String(50), comment="菜单图标")
route_name: Mapped[str | None] = mapped_column(String(100), comment="路由名称")
route_path: Mapped[str | None] = mapped_column(String(200), comment="路由路径")
route_path: Mapped[str | None] = mapped_column(String(200), index=True, comment="路由路径")
component_path: Mapped[str | None] = mapped_column(String(200), comment="组件路径")
redirect: Mapped[str | None] = mapped_column(String(200), comment="重定向地址")
hidden: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False, comment="是否隐藏(True:隐藏 False:显示)")
@@ -46,9 +44,9 @@ class MenuModel(ModelMixin):
show_badge: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False, comment="是否显示红点角标(True:是 False:否)")
show_text_badge: Mapped[str | None] = mapped_column(String(20), comment="文字角标内容")
scope: Mapped[str] = mapped_column(String(20), nullable=False, default="web", server_default="web", comment="菜单可见范围(web:管理端 desktop app:移动端)")
status: Mapped[int] = mapped_column(Integer, default=0, nullable=False, comment="状态(0:启动 1:停用)", index=True)
status: Mapped[int] = mapped_column(Integer, default=0, nullable=False, comment="状态(0:启动 1:停用)")
description: Mapped[str | None] = mapped_column(Text, default=None, nullable=True, comment="备注")
parent_id: Mapped[int | None] = mapped_column(Integer, ForeignKey("sys_menu.id", ondelete="SET NULL"), default=None, index=True, comment="父菜单ID")
parent: Mapped["MenuModel | None"] = relationship(back_populates="children", remote_side="MenuModel.id", foreign_keys="MenuModel.parent_id", uselist=False)
children: Mapped[list["MenuModel"] | None] = relationship(back_populates="parent", foreign_keys="MenuModel.parent_id", order_by="MenuModel.order", lazy="selectin")
roles: Mapped[list["RoleModel"]] = relationship(secondary="sys_role_menus", back_populates="menus", lazy="selectin")
children: Mapped[list["MenuModel"] | None] = relationship(back_populates="parent", foreign_keys="MenuModel.parent_id", order_by="MenuModel.order")
roles: Mapped[list["RoleModel"]] = relationship(secondary="sys_role_menus", back_populates="menus")
backend/app/api/v1/module_system/menu/schema.py
+9 -9
View File
@@ -120,9 +120,9 @@ class MenuUpdateSchema(BaseModel):
active_path: str | None = Field(default=None, max_length=200, description="激活菜单路径")
show_badge: bool | None = Field(default=None, description="是否显示红点角标")
show_text_badge: str | None = Field(default=None, max_length=20, description="文字角标内容")
scope: Literal["platform"] | None = Field(
scope: Literal["web", "app"] | None = Field(
default=None,
description="菜单可见范围",
description="菜单可见范围(web:管理端 app:移动端)",
)
parent_name: str | None = Field(default=None, max_length=50, description="父菜单名称")
@@ -190,13 +190,13 @@ class MenuTreeOutSchema(MenuOutSchema):
class MenuQueryParam(BaseQueryParam):
"""菜单管理查询参数（菜单为平台级资源，无用户归属）"""
name: str | None = Field(None, description="菜单名称")
route_path: str | None = Field(None, description="路由地址")
component_path: str | None = Field(None, description="组件路径")
type: int | None = Field(None, description="菜单类型(1:目录 2:菜单 3:按钮 4:外链)")
permission: str | None = Field(None, description="权限标识")
description: str | None = Field(None, description="描述")
status: int | None = Field(None, description="是否启用")
name: str | None = Field(None, description="菜单名称", json_schema_extra={"q": "like"})
route_path: str | None = Field(None, description="路由地址", json_schema_extra={"q": "like"})
component_path: str | None = Field(None, description="组件路径", json_schema_extra={"q": "like"})
type: int | None = Field(None, description="菜单类型(1:目录 2:菜单 3:按钮 4:外链)", json_schema_extra={"q": "eq"})
permission: str | None = Field(None, description="权限标识", json_schema_extra={"q": "eq"})
description: str | None = Field(None, description="描述", json_schema_extra={"q": "like"})
status: int | None = Field(None, description="是否启用", json_schema_extra={"q": "eq"})
scope: str | None = Field(
None,
description="菜单范围过滤(web:管理端 desktop app:移动端)",
backend/app/api/v1/module_system/menu/service.py
+2 -9
View File
@@ -1,7 +1,6 @@
from typing import Any
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
from app.core.base_schema import AuthSchema, BatchSetAvailable
from app.core.exceptions import CustomException
@@ -19,7 +18,6 @@ from .schema import (
MenuCreateSchema,
MenuOutSchema,
MenuQueryParam,
MenuTreeOutSchema,
MenuUpdateSchema,
)
@@ -77,13 +75,8 @@ class MenuService:
search: MenuQueryParam | None = None,
order_by: list[dict] | None = None,
) -> list[dict]:
# 递归预加载所有层级 children（避免 Pydantic 校验时异步懒加载失败）
from .model import MenuModel
_loader = selectinload(MenuModel.children)
for _ in range(10):
_loader = _loader.selectinload(MenuModel.children)
menu_list = await MenuCRUD(self.auth, self.db).tree_list(search=search_to_dict(search), order_by=order_by, preload=[_loader])
menu_dict_list = [MenuTreeOutSchema.model_validate(menu).model_dump() for menu in menu_list]
menu_list = await MenuCRUD(self.auth, self.db).get_list(search=search_to_dict(search), order_by=order_by)
menu_dict_list = [MenuOutSchema.model_validate(menu).model_dump() for menu in menu_list]
return traversal_to_tree(menu_dict_list)
async def create(self, data: MenuCreateSchema) -> MenuOutSchema:
backend/app/api/v1/module_system/notice/controller.py
+2 -11
View File
@@ -1,9 +1,7 @@
from typing import Annotated
from fastapi import APIRouter, Body, Depends, Path, Query, Security, status
from fastapi import APIRouter, Body, Depends, Path, Security, status
from fastapi.responses import JSONResponse
from fastapi_cache import FastAPICache
from fastapi_cache.decorator import cache
from sqlalchemy.ext.asyncio import AsyncSession
from app.common.response import ResponseSchema, SuccessResponse
@@ -16,8 +14,6 @@ from .service import NoticeService
NoticeRouter = APIRouter(route_class=OperationLogRoute, prefix="/notice", tags=["公告通知"])
_NOTICE_NS = "notice"
@NoticeRouter.get("/detail/{id}", summary="获取公告详情", response_model=ResponseSchema[NoticeOutSchema])
async def get_notice_detail_controller(
@@ -34,7 +30,7 @@ async def get_notice_list_controller(
auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:notice:query"]))],
db: Annotated[AsyncSession, Depends(db_getter)],
page: Annotated[PaginationQueryParam, Depends()],
search: Annotated[NoticeQueryParam, Query()],
search: Annotated[NoticeQueryParam, Depends()],
) -> JSONResponse:
result_dict = await NoticeService(auth, db).page(
page_no=page.page_no,
@@ -52,7 +48,6 @@ async def create_notice_controller(
data: Annotated[NoticeCreateSchema, Body(description="公告创建参数")],
) -> JSONResponse:
result_dict = await NoticeService(auth, db).create(data=data)
await FastAPICache.clear(namespace=_NOTICE_NS)
return SuccessResponse(data=result_dict, msg="创建公告成功")
@@ -64,7 +59,6 @@ async def update_notice_controller(
data: Annotated[NoticeUpdateSchema, Body(description="公告修改参数")],
) -> JSONResponse:
result_dict = await NoticeService(auth, db).update(id=id, data=data)
await FastAPICache.clear(namespace=_NOTICE_NS)
return SuccessResponse(data=result_dict, msg="修改公告成功")
@@ -75,7 +69,6 @@ async def delete_notice_controller(
ids: Annotated[list[int], Body(description="ID列表")],
) -> JSONResponse:
await NoticeService(auth, db).delete(ids=ids)
await FastAPICache.clear(namespace=_NOTICE_NS)
return SuccessResponse(msg="删除公告成功")
@@ -86,12 +79,10 @@ async def batch_set_available_notice_controller(
data: Annotated[BatchSetAvailable, Body(description="状态设置")],
) -> JSONResponse:
await NoticeService(auth, db).set_available(data=data)
await FastAPICache.clear(namespace=_NOTICE_NS)
return SuccessResponse(msg="批量修改公告状态成功")
@NoticeRouter.get("/available", summary="获取全局启用公告", response_model=ResponseSchema[list[NoticeOutSchema]])
@cache(expire=120, namespace=_NOTICE_NS)
async def get_notice_list_available_controller(
auth: Annotated[AuthSchema, Depends(get_current_user)],
db: Annotated[AsyncSession, Depends(db_getter)],
backend/app/api/v1/module_system/notice/model.py
+3 -4
View File
@@ -9,10 +9,9 @@ class NoticeModel(ModelMixin, UserMixin):
__tablename__: str = "sys_notice"
__table_args__: dict[str, str] = {"comment": "通知公告表"}
__loader_options__: list[str] = ["created_by", "updated_by", "deleted_by"]
notice_title: Mapped[str] = mapped_column(String(64), nullable=False, comment="公告标题")
notice_type: Mapped[str] = mapped_column(String(1), nullable=False, comment="公告类型(1通知 2公告)")
notice_title: Mapped[str] = mapped_column(String(64), nullable=False, index=True, comment="公告标题")
notice_type: Mapped[str] = mapped_column(String(1), nullable=False, index=True, comment="公告类型(1通知 2公告)")
notice_content: Mapped[str | None] = mapped_column(Text, nullable=True, comment="公告内容")
status: Mapped[int] = mapped_column(Integer, default=0, nullable=False, comment="状态(0:草稿 1:已发布 2:已归档)", index=True)
status: Mapped[int] = mapped_column(Integer, default=0, nullable=False, comment="状态(0:草稿 1:已发布 2:已归档)")
description: Mapped[str | None] = mapped_column(Text, default=None, nullable=True, comment="备注")
backend/app/api/v1/module_system/notice/schema.py
+2 -2
View File
@@ -62,6 +62,6 @@ class NoticeOutSchema(NoticeCreateSchema, BaseSchema, UserBySchema):
class NoticeQueryParam(BaseQueryParam, UserByQueryParam):
"""公告通知查询参数"""
notice_title: str | None = Field(None, description="公告标题")
notice_title: str | None = Field(None, description="公告标题", json_schema_extra={"q": "like"})
notice_type: str | None = Field(None, description="公告类型", json_schema_extra={"q": "eq"})
status: int | None = Field(None, ge=0, le=2, description="状态(0:草稿 1:已发布 2:已归档)")
status: int | None = Field(None, ge=0, le=2, description="状态(0:草稿 1:已发布 2:已归档)", json_schema_extra={"q": "eq"})
backend/app/api/v1/module_system/notice/service.py
+3 -3
View File
@@ -98,7 +98,7 @@ class NoticeService:
if notice:
raise CustomException(msg="创建失败，该数据已存在")
notice_obj = await NoticeCRUD(self.auth, self.db).create(data=data)
return NoticeOutSchema.model_validate(notice_obj)
return await self.detail(id=notice_obj.id)
async def update(self, id: int, data: NoticeUpdateSchema) -> NoticeOutSchema:
"""更新公告
@@ -114,8 +114,8 @@ class NoticeService:
exist_notice = await NoticeCRUD(self.auth, self.db).get(notice_title=data.notice_title)
if exist_notice and exist_notice.id != id:
raise CustomException(msg="更新失败，标题已存在")
notice_obj = await NoticeCRUD(self.auth, self.db).update(id=id, data=data)
return NoticeOutSchema.model_validate(notice_obj)
await NoticeCRUD(self.auth, self.db).update(id=id, data=data)
return await self.detail(id=id)
async def delete(self, ids: list[int]) -> None:
"""删除公告
backend/app/api/v1/module_system/params/controller.py
+5 -82
View File
@@ -1,59 +1,21 @@
from typing import Annotated
from fastapi import APIRouter, Body, Depends, Path, Query, Security, status
from fastapi.responses import JSONResponse, StreamingResponse
from fastapi import APIRouter, Body, Depends, Path, Security
from fastapi.responses import JSONResponse
from redis.asyncio.client import Redis
from sqlalchemy.ext.asyncio import AsyncSession
from app.common.response import ResponseSchema, StreamResponse, SuccessResponse
from app.core.base_schema import AuthSchema, BatchSetAvailable, PageResultSchema, PaginationQueryParam
from app.common.response import ResponseSchema, SuccessResponse
from app.core.base_schema import AuthSchema
from app.core.dependencies import AuthPermission, db_getter, redis_getter
from app.core.router_class import OperationLogRoute
from app.utils.common_util import bytes2file_response
from .schema import ParamsCreateSchema, ParamsOutSchema, ParamsQueryParam, ParamsUpdateSchema
from .schema import ParamsOutSchema, ParamsUpdateSchema
from .service import ParamsService
ParamsRouter = APIRouter(route_class=OperationLogRoute, prefix="/param", tags=["参数管理"])
@ParamsRouter.get("/detail/{id}", summary="获取参数详情", response_model=ResponseSchema[ParamsOutSchema])
async def get_param_detail_controller(
auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:param:detail"]))],
db: Annotated[AsyncSession, Depends(db_getter)],
id: Annotated[int, Path(description="参数ID", ge=1)],
) -> JSONResponse:
result_dict = await ParamsService(auth, db).detail(id=id)
return SuccessResponse(data=result_dict, msg="获取参数详情成功")
@ParamsRouter.get("/list", summary="获取参数列表", response_model=ResponseSchema[PageResultSchema[ParamsOutSchema]])
async def get_param_list_controller(
auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:param:query"]))],
db: Annotated[AsyncSession, Depends(db_getter)],
page: Annotated[PaginationQueryParam, Depends()],
search: Annotated[ParamsQueryParam, Query()],
) -> JSONResponse:
result_dict = await ParamsService(auth, db).page(
page_no=page.page_no,
page_size=page.page_size,
search=search,
order_by=page.order_by,
)
return SuccessResponse(data=result_dict, msg="查询参数列表成功")
@ParamsRouter.post("/create", status_code=status.HTTP_201_CREATED, summary="创建参数", response_model=ResponseSchema[ParamsOutSchema])
async def create_param_controller(
redis: Annotated[Redis, Depends(redis_getter)],
auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:param:create"]))],
db: Annotated[AsyncSession, Depends(db_getter)],
data: Annotated[ParamsCreateSchema, Body(description="参数创建参数")],
) -> JSONResponse:
result_dict = await ParamsService(auth, db).create(redis=redis, data=data)
return SuccessResponse(data=result_dict, msg="创建参数成功")
@ParamsRouter.put("/update/{id}", summary="修改参数", response_model=ResponseSchema[ParamsOutSchema])
async def update_param_controller(
redis: Annotated[Redis, Depends(redis_getter)],
@@ -66,45 +28,6 @@ async def update_param_controller(
return SuccessResponse(data=result_dict, msg="更新参数成功")
@ParamsRouter.delete("/delete", summary="删除参数", response_model=ResponseSchema[ParamsOutSchema])
async def delete_param_controller(
redis: Annotated[Redis, Depends(redis_getter)],
auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:param:delete"]))],
db: Annotated[AsyncSession, Depends(db_getter)],
ids: Annotated[list[int], Body(description="ID列表")],
) -> JSONResponse:
await ParamsService(auth, db).delete(redis=redis, ids=ids)
return SuccessResponse(msg="删除参数成功")
@ParamsRouter.patch("/status/batch", summary="批量设置参数状态", response_model=ResponseSchema)
async def batch_set_status_controller(
redis: Annotated[Redis, Depends(redis_getter)],
auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:param:patch"]))],
db: Annotated[AsyncSession, Depends(db_getter)],
data: Annotated[BatchSetAvailable, Body(description="状态设置")],
) -> JSONResponse:
await ParamsService(auth, db).batch_set_status(redis=redis, ids=data.ids, status=data.status)
return SuccessResponse(msg="批量设置参数状态成功")
@ParamsRouter.post("/export", summary="导出参数")
async def export_param_list_controller(
auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:param:export"]))],
db: Annotated[AsyncSession, Depends(db_getter)],
search: Annotated[ParamsQueryParam, Body()],
) -> StreamingResponse:
result_dict_list = await ParamsService(auth, db).get_list(search=search)
export_data = [item.model_dump() for item in result_dict_list]
export_result = ParamsService.export(data_list=export_data)
return StreamResponse(
data=bytes2file_response(export_result),
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
headers={"Content-Disposition": "attachment; filename=params.xlsx"},
)
@ParamsRouter.get("/info", summary="获取初始化缓存参数", response_model=ResponseSchema[list[ParamsOutSchema]])
async def get_init_config_controller(
redis: Annotated[Redis, Depends(redis_getter)],
backend/app/api/v1/module_system/params/crud.py
+2 -2
View File
@@ -4,10 +4,10 @@ from app.core.base_crud import CRUDBase
from app.core.base_schema import AuthSchema
from .model import ParamsModel
from .schema import ParamsCreateSchema, ParamsUpdateSchema
from .schema import ParamsUpdateSchema
class ParamsCRUD(CRUDBase[ParamsModel, ParamsCreateSchema, ParamsUpdateSchema]):
class ParamsCRUD(CRUDBase[ParamsModel, ParamsUpdateSchema, ParamsUpdateSchema]):
"""配置管理数据层"""
def __init__(self, auth: AuthSchema, db: AsyncSession) -> None:
backend/app/api/v1/module_system/params/model.py
+5 -6
View File
@@ -1,19 +1,18 @@
from sqlalchemy import Boolean, Integer, String, Text
from sqlalchemy.orm import Mapped, mapped_column
from app.core.base_model import ModelMixin, UserMixin
from app.core.base_model import ModelMixin
class ParamsModel(ModelMixin, UserMixin):
class ParamsModel(ModelMixin):
"""系统参数表"""
__tablename__: str = "sys_param"
__table_args__: dict[str, str] = {"comment": "系统参数表"}
__loader_options__: list[str] = ["created_by", "updated_by", "deleted_by"]
config_name: Mapped[str] = mapped_column(String(64), nullable=False, comment="参数名称")
config_key: Mapped[str] = mapped_column(String(500), nullable=False, comment="参数键名")
config_name: Mapped[str] = mapped_column(String(64), nullable=False, index=True, comment="参数名称")
config_key: Mapped[str] = mapped_column(String(500), nullable=False, index=True, comment="参数键名")
config_value: Mapped[str | None] = mapped_column(Text, comment="参数键值")
config_type: Mapped[bool] = mapped_column(Boolean, default=False, nullable=True, comment="系统内置(True:是 False:否)", index=True)
status: Mapped[int] = mapped_column(Integer, default=0, nullable=False, comment="状态(0:启动 1:停用)", index=True)
status: Mapped[int] = mapped_column(Integer, default=0, nullable=False, comment="状态(0:启动 1:停用)")
description: Mapped[str | None] = mapped_column(Text, default=None, nullable=True, comment="备注")
backend/app/api/v1/module_system/params/schema.py
+5 -15
View File
@@ -2,11 +2,11 @@ import re
from pydantic import BaseModel, ConfigDict, Field, field_validator
from app.core.base_schema import BaseQueryParam, BaseSchema, UserByQueryParam, UserBySchema
from app.core.base_schema import BaseSchema
class ParamsCreateSchema(BaseModel):
"""参数创建模型
class ParamsBaseSchema(BaseModel):
"""参数基础字段
"""
config_name: str = Field(..., min_length=1, max_length=64, description="参数名称")
@@ -34,23 +34,13 @@ class ParamsCreateSchema(BaseModel):
return v
class ParamsUpdateSchema(ParamsCreateSchema):
class ParamsUpdateSchema(ParamsBaseSchema):
"""参数更新模型
"""
class ParamsOutSchema(ParamsCreateSchema, BaseSchema, UserBySchema):
class ParamsOutSchema(ParamsBaseSchema, BaseSchema):
"""参数响应模型
"""
model_config = ConfigDict(from_attributes=True)
class ParamsQueryParam(BaseQueryParam, UserByQueryParam):
"""参数管理查询参数
"""
config_name: str | None = Field(None, description="参数名称")
config_key: str | None = Field(None, description="参数键名", json_schema_extra={"q": "eq"})
config_type: bool | None = Field(None, description="是否系统内置(True:是 False:否)")
status: int | None = Field(None, ge=0, le=1, description="状态(0:启动 1:停用)")
backend/app/api/v1/module_system/params/service.py
+3 -216
View File
@@ -5,144 +5,23 @@ from redis.asyncio.client import Redis
from sqlalchemy.ext.asyncio import AsyncSession
from app.common.enums import RedisInitKeyConfig
from app.core.base_schema import AuthSchema, PageResultSchema
from app.core.base_schema import AuthSchema
from app.core.database import async_db_session
from app.core.exceptions import CustomException
from app.core.logger import logger
from app.core.redis_crud import RedisCURD
from app.utils.common_util import search_to_dict
from app.utils.excel_util import ExcelUtil
from .crud import ParamsCRUD
from .schema import (
ParamsCreateSchema,
ParamsOutSchema,
ParamsQueryParam,
ParamsUpdateSchema,
)
from .schema import ParamsOutSchema, ParamsUpdateSchema
class ParamsService:
"""参数管理服务
设计：实例方法承载「当前用户上下文 (auth)」，``redis`` 仍是方法参数
（因为不是每个端点都用到）。调用方写法由
``ParamsService.method_service(auth=...)`` 改为 ``ParamsService(auth).method(...)``。
"""
"""参数管理服务"""
def __init__(self, auth: AuthSchema, db: AsyncSession) -> None:
self.auth = auth
self.db = db
async def detail(self, id: int) -> ParamsOutSchema:
"""获取参数详情
参数:
- id (int): 参数ID
返回:
- ParamsOutSchema: 参数响应模型
"""
obj = await ParamsCRUD(self.auth, self.db).get_or_404(id=id)
return ParamsOutSchema.model_validate(obj)
async def get_by_key(self, config_key: str) -> ParamsOutSchema:
"""根据配置键获取参数详情
参数:
- config_key (str): 参数键名
返回:
- ParamsOutSchema: 参数响应模型
"""
obj = await ParamsCRUD(self.auth, self.db).get(config_key=config_key)
if not obj:
raise CustomException(msg="该数据不存在")
return ParamsOutSchema.model_validate(obj)
async def get_list(
self,
search: ParamsQueryParam | None = None,
order_by: list[dict] | None = None,
) -> list[ParamsOutSchema]:
"""获取配置管理型列表
参数:
- search (ParamsQueryParam | None): 查询参数对象
- order_by (list[dict] | None): 排序参数列表
返回:
- list[ParamsOutSchema]: 参数响应模型列表
"""
obj_list = await ParamsCRUD(self.auth, self.db).get_list(search=search_to_dict(search), order_by=order_by)
return [ParamsOutSchema.model_validate(obj) for obj in obj_list]
async def page(
self,
page_no: int,
page_size: int,
search: ParamsQueryParam | None = None,
order_by: list[dict[str, str]] | None = None,
) -> PageResultSchema[ParamsOutSchema]:
"""分页查询系统参数（数据库 OFFSET/LIMIT）。
参数:
- page_no (int): 页码（从 1 开始）
- page_size (int): 每页条数
- search (ParamsQueryParam | None): 查询条件
- order_by (list[dict[str, str]] | None): 排序字段列表
返回:
- PageResultSchema[ParamsOutSchema]: 分页结果
"""
offset = (page_no - 1) * page_size
return await ParamsCRUD(self.auth, self.db).page(
offset=offset,
limit=page_size,
order_by=order_by or [{"id": "asc"}],
search=search_to_dict(search),
out_schema=ParamsOutSchema,
)
async def create(self, redis: Redis, data: ParamsCreateSchema) -> ParamsOutSchema:
"""创建配置管理型
参数:
- redis (Redis): Redis 客户端实例
- data (ParamsCreateSchema): 配置管理型创建模型
返回:
- ParamsOutSchema: 新创建的参数响应模型
"""
exist_obj = await ParamsCRUD(self.auth, self.db).get(config_key=data.config_key)
if exist_obj:
raise CustomException(msg="创建失败，该数据已存在")
obj = await ParamsCRUD(self.auth, self.db).create(data=data)
out = ParamsOutSchema.model_validate(obj)
# 同步redis
user = self.auth.user
if not user:
raise CustomException(msg="未登录")
redis_key = f"{RedisInitKeyConfig.SYSTEM_CONFIG.key}:{data.config_key}"
try:
redis_payload = out.model_dump(mode="json")
value = json.dumps(redis_payload, ensure_ascii=False)
result = await RedisCURD(redis).set(
key=redis_key,
value=value,
expire=None,
)
if not result:
logger.error(f"同步配置到缓存失败: {out}")
raise CustomException(msg="同步配置到缓存失败")
except Exception as e:
logger.error(f"创建字典类型失败: {e}")
raise CustomException(msg="同步配置到缓存失败") from e
return out
async def update(self, redis: Redis, id: int, data: ParamsUpdateSchema) -> ParamsOutSchema:
"""更新参数
@@ -185,98 +64,6 @@ class ParamsService:
return out
async def delete(self, redis: Redis, ids: list[int]) -> None:
"""删除配置管理型
参数:
- redis (Redis): Redis 客户端实例
- ids (list[int]): 配置管理型ID列表
返回:
- None
"""
if not ids:
raise CustomException(msg="删除失败，删除对象不能为空")
# 批量校验参数存在性
objs = await ParamsCRUD(self.auth, self.db).get_list(search={"id": ("in", ids)})
obj_map = {o.id: o for o in objs}
for pid in ids:
obj = obj_map.get(pid)
if not obj:
raise CustomException(msg="删除失败，该数据不存在")
if obj.config_type:
raise CustomException(msg=f"{obj.config_name} 删除失败，系统初始化配置不可以删除")
await ParamsCRUD(self.auth, self.db).delete(ids=ids)
# 同步删除Redis缓存（使用删除前已获取的对象信息）
user = self.auth.user
if not user:
raise CustomException(msg="未登录")
for obj in objs:
redis_key = f"{RedisInitKeyConfig.SYSTEM_CONFIG.key}:{obj.config_key}"
try:
await RedisCURD(redis).delete(redis_key)
except Exception as e:
logger.error(f"删除系统配置失败: {e}")
raise CustomException(msg="同步删除缓存失败") from e
async def batch_set_status(self, redis: Redis, ids: list[int], status: int) -> None:
"""批量设置系统参数状态
参数:
- redis: Redis 客户端（用于同步缓存）
- ids (list[int]): 系统参数ID列表
- status (int): 状态值
返回:
- None
"""
if not ids:
raise CustomException(msg="请选择要操作的数据")
# 先查参数列表获取 config_key
params = await ParamsCRUD(self.auth, self.db).get_list(search={"id": ("in", list(ids))})
await ParamsCRUD(self.auth, self.db).set(ids=ids, status=status)
# 同步删除对应 Redis 缓存
for param in params:
redis_key = f"{RedisInitKeyConfig.SYSTEM_CONFIG.key}:{param.config_key}"
try:
await RedisCURD(redis).delete(redis_key)
except Exception as e:
logger.error(f"同步删除系统配置缓存失败: {e}")
@staticmethod
def export(data_list: list[dict]) -> bytes:
"""导出参数列表（无状态工具方法）
参数:
- data_list (list[dict]): 参数字典列表
返回:
- bytes: Excel 文件字节流
"""
mapping_dict = {
"id": "编号",
"config_name": "参数名称",
"config_key": "参数键名",
"config_value": "参数键值",
"config_type": "系统内置((True:是 False:否))",
"description": "备注",
"created_time": "创建时间",
"updated_time": "更新时间",
"created_id": "创建者ID",
"updated_id": "更新者ID",
}
# 复制数据并转换状态
data = data_list.copy()
for item in data:
# 处理状态
item["config_type"] = "是" if item.get("config_type") else "否"
return ExcelUtil.export_list2excel(list_data=data, mapping_dict=mapping_dict)
@staticmethod
async def _load_all_configs_from_db() -> Sequence[object]:
async with async_db_session() as session, session.begin():
backend/app/api/v1/module_system/position/controller.py
+2 -11
View File
@@ -1,9 +1,7 @@
from typing import Annotated
from fastapi import APIRouter, Body, Depends, Path, Query, Security, status
from fastapi import APIRouter, Body, Depends, Path, Security, status
from fastapi.responses import JSONResponse, StreamingResponse
from fastapi_cache import FastAPICache
from fastapi_cache.decorator import cache
from sqlalchemy.ext.asyncio import AsyncSession
from app.common.response import ResponseSchema, StreamResponse, SuccessResponse
@@ -17,16 +15,13 @@ from .service import PositionService
PositionRouter = APIRouter(route_class=OperationLogRoute, prefix="/position", tags=["岗位管理"])
_POS_NS = "position"
@PositionRouter.get("/list", summary="查询岗位", response_model=ResponseSchema[PageResultSchema[PositionOutSchema]])
@cache(expire=300, namespace=_POS_NS)
async def get_obj_list_controller(
auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:position:query"]))],
db: Annotated[AsyncSession, Depends(db_getter)],
page: Annotated[PaginationQueryParam, Depends()],
search: Annotated[PositionQueryParam, Query()],
search: Annotated[PositionQueryParam, Depends()],
) -> JSONResponse:
result_dict = await PositionService(auth, db).page(
page_no=page.page_no,
@@ -54,7 +49,6 @@ async def create_obj_controller(
data: Annotated[PositionCreateSchema, Body(description="岗位创建参数")],
) -> JSONResponse:
result_dict = await PositionService(auth, db).create(data=data)
await FastAPICache.clear(namespace=_POS_NS)
return SuccessResponse(data=result_dict, msg="创建岗位成功")
@@ -66,7 +60,6 @@ async def update_obj_controller(
data: Annotated[PositionUpdateSchema, Body(description="岗位修改参数")],
) -> JSONResponse:
result_dict = await PositionService(auth, db).update(id=id, data=data)
await FastAPICache.clear(namespace=_POS_NS)
return SuccessResponse(data=result_dict, msg="修改岗位成功")
@@ -77,7 +70,6 @@ async def delete_obj_controller(
ids: Annotated[list[int], Body(description="ID列表")],
) -> JSONResponse:
await PositionService(auth, db).delete(ids=ids)
await FastAPICache.clear(namespace=_POS_NS)
return SuccessResponse(msg="删除岗位成功")
@@ -88,7 +80,6 @@ async def batch_set_available_obj_controller(
data: Annotated[BatchSetAvailable, Body(description="状态设置")],
) -> JSONResponse:
await PositionService(auth, db).set_available(data=data)
await FastAPICache.clear(namespace=_POS_NS)
return SuccessResponse(msg="批量修改岗位状态成功")
backend/app/api/v1/module_system/position/model.py
+3 -8
View File
@@ -14,17 +14,12 @@ class PositionModel(ModelMixin, UserMixin):
__tablename__: str = "sys_position"
__table_args__: dict[str, str] = {"comment": "岗位表"}
__loader_options__: list[str] = ["users", "created_by", "updated_by", "deleted_by"]
name: Mapped[str] = mapped_column(String(64), nullable=False, comment="岗位名称")
name: Mapped[str] = mapped_column(String(64), nullable=False, index=True, comment="岗位名称")
code: Mapped[str] = mapped_column(String(64), unique=True, nullable=False, comment="岗位编码")
order: Mapped[int] = mapped_column(Integer, nullable=False, default=1, comment="显示排序")
status: Mapped[int] = mapped_column(Integer, default=0, nullable=False, comment="状态(0:启动 1:停用)", index=True)
status: Mapped[int] = mapped_column(Integer, default=0, nullable=False, comment="状态(0:启动 1:停用)")
description: Mapped[str | None] = mapped_column(Text, default=None, nullable=True, comment="备注")
# 关联关系
users: Mapped[list["UserModel"]] = relationship(
secondary="sys_user_positions",
back_populates="positions",
lazy="selectin",
)
users: Mapped[list["UserModel"]] = relationship(secondary="sys_user_positions", back_populates="positions")
backend/app/api/v1/module_system/position/schema.py
+2 -2
View File
@@ -49,5 +49,5 @@ class PositionOutSchema(PositionCreateSchema, BaseSchema, UserBySchema):
class PositionQueryParam(BaseQueryParam, UserByQueryParam):
"""岗位管理查询参数"""
name: str | None = Field(None, description="岗位名称")
status: int | None = Field(None, ge=0, le=1, description="状态(0:启动 1:停用)")
name: str | None = Field(None, description="岗位名称", json_schema_extra={"q": "like"})
status: int | None = Field(None, ge=0, le=1, description="状态(0:启动 1:停用)", json_schema_extra={"q": "eq"})
backend/app/api/v1/module_system/position/service.py
+3 -3
View File
@@ -63,15 +63,15 @@ class PositionService:
if position:
raise CustomException(msg="创建失败，该数据已存在")
new_position = await PositionCRUD(self.auth, self.db).create(data=data)
return PositionOutSchema.model_validate(new_position)
return await self.detail(id=new_position.id)
async def update(self, id: int, data: PositionUpdateSchema) -> PositionOutSchema:
_ = await PositionCRUD(self.auth, self.db).get_or_404(id=id, msg="更新失败，该数据不存在")
exist_position = await PositionCRUD(self.auth, self.db).get(name=data.name)
if exist_position and exist_position.id != id:
raise CustomException(msg="更新失败，名称已存在")
updated_position = await PositionCRUD(self.auth, self.db).update(id=id, data=data)
return PositionOutSchema.model_validate(updated_position)
await PositionCRUD(self.auth, self.db).update(id=id, data=data)
return await self.detail(id=id)
async def delete(self, ids: list[int]) -> None:
if not ids:
backend/app/api/v1/module_system/role/controller.py
+2 -12
View File
@@ -1,9 +1,7 @@
from typing import Annotated
from fastapi import APIRouter, Body, Depends, Path, Query, Security, status
from fastapi import APIRouter, Body, Depends, Path, Security, status
from fastapi.responses import JSONResponse, StreamingResponse
from fastapi_cache import FastAPICache
from fastapi_cache.decorator import cache
from sqlalchemy.ext.asyncio import AsyncSession
from app.common.response import ResponseSchema, StreamResponse, SuccessResponse
@@ -17,16 +15,13 @@ from .service import RoleService
RoleRouter = APIRouter(route_class=OperationLogRoute, prefix="/role", tags=["角色管理"])
_ROLE_NS = "role"
@RoleRouter.get("/list", summary="查询角色", response_model=ResponseSchema[PageResultSchema[RoleOutSchema]])
@cache(expire=300, namespace=_ROLE_NS)
async def get_role_list_controller(
auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:role:query"]))],
db: Annotated[AsyncSession, Depends(db_getter)],
page: Annotated[PaginationQueryParam, Depends()],
search: Annotated[RoleQueryParam, Query()],
search: Annotated[RoleQueryParam, Depends()],
) -> JSONResponse:
result_dict = await RoleService(auth, db).page(
page_no=page.page_no,
@@ -54,7 +49,6 @@ async def create_role_controller(
data: Annotated[RoleCreateSchema, Body(description="角色创建参数")],
) -> JSONResponse:
result_dict = await RoleService(auth, db).create(data=data)
await FastAPICache.clear(namespace=_ROLE_NS)
return SuccessResponse(data=result_dict, msg="创建角色成功")
@@ -66,7 +60,6 @@ async def update_role_controller(
data: Annotated[RoleUpdateSchema, Body(description="角色修改参数")],
) -> JSONResponse:
result_dict = await RoleService(auth, db).update(id=id, data=data)
await FastAPICache.clear(namespace=_ROLE_NS)
return SuccessResponse(data=result_dict, msg="修改角色成功")
@@ -77,7 +70,6 @@ async def delete_role_controller(
ids: Annotated[list[int], Body(description="ID列表")],
) -> JSONResponse:
await RoleService(auth, db).delete(ids=ids)
await FastAPICache.clear(namespace=_ROLE_NS)
return SuccessResponse(msg="删除角色成功")
@@ -88,7 +80,6 @@ async def batch_set_available_role_controller(
data: Annotated[BatchSetAvailable, Body(description="状态设置")],
) -> JSONResponse:
await RoleService(auth, db).set_available(data=data)
await FastAPICache.clear(namespace=_ROLE_NS)
return SuccessResponse(msg="批量修改角色状态成功")
@@ -99,7 +90,6 @@ async def set_role_permission_controller(
data: Annotated[RolePermissionSettingSchema, Body(description="角色授权参数")],
) -> JSONResponse:
await RoleService(auth, db).set_permission(data=data)
await FastAPICache.clear(namespace=_ROLE_NS)
return SuccessResponse(msg="授权角色成功")
backend/app/api/v1/module_system/role/crud.py
+2 -4
View File
@@ -31,15 +31,13 @@ class RoleCRUD(CRUDBase[RoleModel, RoleCreateSchema, RoleUpdateSchema]):
if not role_ids:
raise CustomException(msg="角色ID列表不能为空")
roles = await self.get_list(search={"id": ("in", role_ids)})
# 校验：传入的 role_ids 必须全部存在（否则容易被 IDOR silent no-op）
roles = await self.get_list(search={"id": ("in", role_ids)}, preload=["menus"])
if len(roles) != len(set(role_ids)):
missing = sorted(set(role_ids) - {r.id for r in roles})
raise CustomException(msg=f"角色不存在: {missing}")
menus = [] if not menu_ids else await MenuCRUD(self.auth, self.db).get_list(search={"id": ("in", menu_ids)})
# 校验：传入的所有菜单必须存在
if menu_ids and len(menus) != len(set(menu_ids)):
missing = sorted(set(menu_ids) - {m.id for m in menus})
raise CustomException(msg=f"菜单不存在: {missing}")
@@ -54,7 +52,7 @@ class RoleCRUD(CRUDBase[RoleModel, RoleCreateSchema, RoleUpdateSchema]):
if not role_ids:
raise CustomException(msg="角色ID列表不能为空")
roles = await self.get_list(search={"id": ("in", role_ids)})
roles = await self.get_list(search={"id": ("in", role_ids)}, preload=["depts"])
if len(roles) != len(set(role_ids)):
missing = sorted(set(role_ids) - {r.id for r in roles})
raise CustomException(msg=f"角色不存在: {missing}")
backend/app/api/v1/module_system/role/model.py
+5 -11
View File
@@ -63,20 +63,14 @@ class RoleModel(ModelMixin, UserMixin):
__tablename__: str = "sys_role"
__table_args__: dict[str, str] = {"comment": "角色表"}
__loader_options__: list[str] = ["menus", "depts", "created_by", "updated_by", "deleted_by"]
name: Mapped[str] = mapped_column(String(64), nullable=False, comment="角色名称")
name: Mapped[str] = mapped_column(String(64), nullable=False, index=True, comment="角色名称")
code: Mapped[str] = mapped_column(String(64), unique=True, nullable=False, comment="角色编码")
order: Mapped[int] = mapped_column(Integer, nullable=False, default=999, comment="显示排序")
status: Mapped[int] = mapped_column(Integer, default=0, nullable=False, comment="状态(0:启动 1:停用)", index=True)
status: Mapped[int] = mapped_column(Integer, default=0, nullable=False, comment="状态(0:启动 1:停用)")
description: Mapped[str | None] = mapped_column(Text, default=None, nullable=True, comment="备注")
data_scope: Mapped[int] = mapped_column(Integer, default=1, nullable=False, comment="数据权限范围(1:仅本人 2:本部门 3:本部门及以下 4:全部 5:自定义)")
menus: Mapped[list["MenuModel"]] = relationship(
secondary="sys_role_menus",
back_populates="roles",
lazy="selectin",
order_by="MenuModel.order",
)
depts: Mapped[list["DeptModel"]] = relationship(secondary="sys_role_depts", back_populates="roles", lazy="selectin")
users: Mapped[list["UserModel"]] = relationship(secondary="sys_user_roles", back_populates="roles", lazy="selectin")
menus: Mapped[list["MenuModel"]] = relationship(secondary="sys_role_menus", back_populates="roles", order_by="MenuModel.order")
depts: Mapped[list["DeptModel"]] = relationship(secondary="sys_role_depts", back_populates="roles")
users: Mapped[list["UserModel"]] = relationship(secondary="sys_user_roles", back_populates="roles")
backend/app/api/v1/module_system/role/schema.py
+2 -2
View File
@@ -98,6 +98,6 @@ class RoleQueryParam(BaseQueryParam, UserByQueryParam):
"""角色管理查询参数
"""
name: str | None = Field(None, description="角色名称")
name: str | None = Field(None, description="角色名称", json_schema_extra={"q": "like"})
code: str | None = Field(None, description="角色编码", json_schema_extra={"q": "eq"})
status: int | None = Field(None, description="状态(0:启动 1:停用)")
status: int | None = Field(None, description="状态(0:启动 1:停用)", json_schema_extra={"q": "eq"})
backend/app/api/v1/module_system/role/service.py
+8 -5
View File
@@ -16,6 +16,8 @@ from .schema import (
RoleUpdateSchema,
)
_ROLE_PRELOAD = ["menus", "depts"]
class RoleService:
"""角色管理服务
@@ -36,7 +38,7 @@ class RoleService:
返回:
- RoleOutSchema: 角色详情响应模型
"""
obj = await RoleCRUD(self.auth, self.db).get_or_404(id=id)
obj = await RoleCRUD(self.auth, self.db).get_or_404(id=id, preload=_ROLE_PRELOAD)
return RoleOutSchema.model_validate(obj)
async def get_options(self) -> list[dict[str, Any]]:
@@ -57,7 +59,7 @@ class RoleService:
返回:
- list[RoleOutSchema]: 角色响应模型列表
"""
role_list = await RoleCRUD(self.auth, self.db).get_list(search=search_to_dict(search), order_by=order_by)
role_list = await RoleCRUD(self.auth, self.db).get_list(search=search_to_dict(search), order_by=order_by, preload=_ROLE_PRELOAD)
return [RoleOutSchema.model_validate(role) for role in role_list]
async def page(
@@ -85,6 +87,7 @@ class RoleService:
order_by=order_by or [{"id": "asc"}],
search=search_to_dict(search),
out_schema=RoleOutSchema,
preload=_ROLE_PRELOAD,
)
async def create(self, data: RoleCreateSchema) -> RoleOutSchema:
@@ -105,7 +108,7 @@ class RoleService:
raise CustomException(msg="创建失败，编码已存在")
new_role = await RoleCRUD(self.auth, self.db).create(data=data)
return RoleOutSchema.model_validate(new_role)
return await self.detail(id=new_role.id)
async def update(self, id: int, data: RoleUpdateSchema) -> RoleOutSchema:
"""更新角色
@@ -124,8 +127,8 @@ class RoleService:
exist_code = await RoleCRUD(self.auth, self.db).get(code=data.code)
if exist_code and exist_code.id != id:
raise CustomException(msg="更新失败，角色编码已存在")
updated_role = await RoleCRUD(self.auth, self.db).update(id=id, data=data)
return RoleOutSchema.model_validate(updated_role)
await RoleCRUD(self.auth, self.db).update(id=id, data=data)
return await self.detail(id=id)
async def delete(self, ids: list[int]) -> None:
"""删除角色
backend/app/api/v1/module_system/ticket/controller.py
+2 -2
View File
@@ -1,6 +1,6 @@
from typing import Annotated
from fastapi import APIRouter, Body, Depends, Path, Query, Security, status
from fastapi import APIRouter, Body, Depends, Path, Security, status
from fastapi.responses import JSONResponse, StreamingResponse
from sqlalchemy.ext.asyncio import AsyncSession
@@ -21,7 +21,7 @@ async def ticket_list_controller(
auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:ticket:query"]))],
db: Annotated[AsyncSession, Depends(db_getter)],
page: Annotated[PaginationQueryParam, Depends()],
search: Annotated[TicketQueryParam, Query()],
search: Annotated[TicketQueryParam, Depends()],
) -> JSONResponse:
result = await TicketService(auth, db).page(
page_no=page.page_no,
backend/app/api/v1/module_system/ticket/model.py
+3 -14
View File
@@ -16,15 +16,9 @@ class TicketModel(ModelMixin, UserMixin):
__tablename__: str = "sys_ticket"
__table_args__: dict[str, str] = {"comment": "工单表"}
__loader_options__: list[str] = [
"created_by",
"updated_by",
"deleted_by",
"assigned_by",
]
title: Mapped[str] = mapped_column(String(200), nullable=False, comment="工单标题")
status: Mapped[int] = mapped_column(Integer, default=0, nullable=False, comment="状态(0:待处理 1:处理中 2:已完成 3:已关闭)", index=True)
title: Mapped[str] = mapped_column(String(200), nullable=False, index=True, comment="标题")
status: Mapped[int] = mapped_column(Integer, default=0, nullable=False, comment="状态(0:待处理 1:处理中 2:已完成 3:已关闭)")
description: Mapped[str | None] = mapped_column(Text, default=None, nullable=True, comment="备注")
ticket_content: Mapped[str | None] = mapped_column(Text, nullable=True, comment="工单内容（富文本）")
summary: Mapped[str | None] = mapped_column(Text, nullable=True, comment="工单内容（纯文本摘要）")
@@ -33,7 +27,7 @@ class TicketModel(ModelMixin, UserMixin):
reply: Mapped[str | None] = mapped_column(Text, nullable=True, comment="回复内容")
assigned_id: Mapped[int | None] = mapped_column(ForeignKey("sys_user.id", ondelete="SET NULL", onupdate="CASCADE"), nullable=True, index=True, comment="处理人ID")
assigned_by: Mapped["UserModel | None"] = relationship("UserModel", foreign_keys=[assigned_id], lazy="selectin", uselist=False)
assigned_by: Mapped["UserModel | None"] = relationship("UserModel", foreign_keys=[assigned_id], uselist=False)
@validates("title")
def validate_title(self, key: str, title: str) -> str:
@@ -52,11 +46,6 @@ class TicketCommentModel(ModelMixin, UserMixin):
"""工单评论模型"""
__tablename__: str = "sys_ticket_comment"
__table_args__: dict[str, str] = {"comment": "工单评论表"}
__loader_options__: list[str] = [
"created_by",
"updated_by",
"deleted_by",
]
ticket_id: Mapped[int] = mapped_column(ForeignKey("sys_ticket.id", ondelete="CASCADE"), nullable=False, index=True, comment="工单ID")
content: Mapped[str] = mapped_column(Text, nullable=False, comment="评论内容（富文本）")
backend/app/api/v1/module_system/ticket/schema.py
+3 -3
View File
@@ -84,10 +84,10 @@ class TicketBatchSchema(BaseModel):
class TicketQueryParam(BaseQueryParam, UserByQueryParam):
"""工单查询参数"""
title: str | None = Field(None, description="工单标题")
title: str | None = Field(None, description="工单标题", json_schema_extra={"q": "like"})
ticket_type: str | None = Field(None, description="工单类型", json_schema_extra={"q": "eq"})
assigned_id: int | None = Field(None, description="处理人ID")
status: int | None = Field(None, ge=0, le=3, description="状态(0:待处理 1:处理中 2:已完成 3:已关闭)")
assigned_id: int | None = Field(None, description="处理人ID", json_schema_extra={"q": "eq"})
status: int | None = Field(None, ge=0, le=3, description="状态(0:待处理 1:处理中 2:已完成 3:已关闭)", json_schema_extra={"q": "eq"})
class TicketCommentCreateSchema(BaseModel):
backend/app/api/v1/module_system/ticket/service.py
+8 -16
View File
@@ -5,7 +5,6 @@ from sqlalchemy.ext.asyncio import AsyncSession
from app.api.v1.module_system.user.model import UserModel
from app.core.base_schema import AuthSchema, PageResultSchema
from app.core.event_bus import EventBus
from app.core.exceptions import CustomException
from app.utils.common_util import search_to_dict
from app.utils.excel_util import ExcelUtil
@@ -36,6 +35,9 @@ _TICKET_STATUS_LABELS = {
}
_TICKET_PRELOAD = ["assigned_by"]
class TicketService:
"""工单管理服务"""
@@ -88,17 +90,18 @@ class TicketService:
order_by=order_by or [{"created_time": "desc"}],
search=search_to_dict(search),
out_schema=TicketOutSchema,
preload=_TICKET_PRELOAD,
)
async def detail(self, id: int) -> TicketOutSchema:
obj = await TicketCRUD(self.auth, self.db).get_or_404(id=id)
obj = await TicketCRUD(self.auth, self.db).get_or_404(id=id, preload=_TICKET_PRELOAD)
return TicketOutSchema.model_validate(obj)
async def create(self, data: TicketCreateSchema) -> TicketOutSchema:
obj = await TicketCRUD(self.auth, self.db).create(data=data)
if not obj:
raise CustomException(msg="创建工单失败")
return TicketOutSchema.model_validate(obj)
return await self.detail(id=obj.id)
async def update(self, id: int, data: TicketUpdateSchema) -> TicketOutSchema:
obj = await TicketCRUD(self.auth, self.db).get_or_404(id=id, msg="工单不存在")
@@ -120,19 +123,7 @@ class TicketService:
if not updated:
raise CustomException(msg="工单不存在")
# 有回复内容时 SSE 推送通知给工单创建者
if data.reply and obj.created_id:
await EventBus.publish(
obj.created_id,
{
"type": "ticket_reply",
"ticket_id": obj.id,
"title": obj.title,
"ticket_type": obj.ticket_type,
},
)
return TicketOutSchema.model_validate(updated)
return await self.detail(id=updated.id)
async def delete(self, ids: list[int]) -> None:
if not ids:
@@ -160,6 +151,7 @@ class TicketService:
obj_list = await TicketCRUD(self.auth, self.db).get_list(
search=search_to_dict(search),
order_by=order_by or [{"created_time": "desc"}],
preload=_TICKET_PRELOAD,
)
return [TicketOutSchema.model_validate(obj) for obj in obj_list]
backend/app/api/v1/module_system/user/controller.py
+2 -2
View File
@@ -1,7 +1,7 @@
import urllib.parse
from typing import Annotated
from fastapi import APIRouter, Body, Depends, File, Path, Query, Security, UploadFile, status
from fastapi import APIRouter, Body, Depends, File, Path, Security, UploadFile, status
from fastapi.responses import JSONResponse, StreamingResponse
from redis.asyncio.client import Redis
from sqlalchemy.ext.asyncio import AsyncSession
@@ -120,7 +120,7 @@ async def get_user_list_controller(
auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:user:query"]))],
db: Annotated[AsyncSession, Depends(db_getter)],
page: Annotated[PaginationQueryParam, Depends()],
search: Annotated[UserQueryParam, Query()],
search: Annotated[UserQueryParam, Depends()],
) -> JSONResponse:
result_dict = await UserService(auth, db).page(
page_no=page.page_no,
backend/app/api/v1/module_system/user/crud.py
+8 -26
View File
@@ -37,41 +37,23 @@ class UserCRUD(CRUDBase[UserModel, UserCreateSchema, UserUpdateSchema]):
await self.set([id], last_login=datetime.now())
async def set_user_roles(self, user_ids: list[int], role_ids: list[int]) -> None:
"""批量设置用户角色
参数:
- user_ids (list[int]): 用户ID列表
- role_ids (list[int]): 角色ID列表
返回:
- None
"""
user_objs = await self.get_list(search={"id": ("in", user_ids)})
"""批量设置用户角色"""
user_objs = await self.get_list(search={"id": ("in", user_ids)}, preload=["roles"])
role_objs = [] if not role_ids else await RoleCRUD(self.auth, self.db).get_list(search={"id": ("in", role_ids)})
for obj in user_objs:
relationship = obj.roles
relationship.clear()
relationship.extend(role_objs)
obj.roles.clear()
obj.roles.extend(role_objs)
await self.db.flush()
async def set_user_positions(self, user_ids: list[int], position_ids: list[int]) -> None:
"""批量设置用户岗位
参数:
- user_ids (list[int]): 用户ID列表
- position_ids (list[int]): 岗位ID列表
返回:
- None
"""
user_objs = await self.get_list(search={"id": ("in", user_ids)})
"""批量设置用户岗位"""
user_objs = await self.get_list(search={"id": ("in", user_ids)}, preload=["positions"])
position_objs = [] if not position_ids else await PositionCRUD(self.auth, self.db).get_list(search={"id": ("in", position_ids)})
for obj in user_objs:
relationship = obj.positions
relationship.clear()
relationship.extend(position_objs)
obj.positions.clear()
obj.positions.extend(position_objs)
await self.db.flush()
async def change_password(self, id: int, password_hash: str) -> UserModel:
backend/app/api/v1/module_system/user/model.py
+9 -10
View File
@@ -64,13 +64,12 @@ class UserModel(ModelMixin, UserMixin):
__tablename__: str = "sys_user"
__table_args__: dict[str, str] = {"comment": "用户表"}
__loader_options__: list[str] = ["dept", "roles", "positions", "created_by", "updated_by", "deleted_by"]
username: Mapped[str] = mapped_column(String(64), unique=True, nullable=False, comment="用户名/登录账号")
password: Mapped[str] = mapped_column(String(255), nullable=False, comment="密码哈希")
name: Mapped[str] = mapped_column(String(32), nullable=False, comment="昵称")
mobile: Mapped[str | None] = mapped_column(String(11), nullable=True, comment="手机号")
email: Mapped[str | None] = mapped_column(String(64), nullable=True, comment="邮箱")
mobile: Mapped[str | None] = mapped_column(String(11), nullable=True, index=True, comment="手机号")
email: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True, comment="邮箱")
gender: Mapped[str | None] = mapped_column(String(1), default="2", nullable=True, comment="性别(0:男 1:女 2:未知)")
avatar: Mapped[str | None] = mapped_column(String(255), nullable=True, comment="头像URL地址")
is_superuser: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False, comment="是否超管")
@@ -79,13 +78,13 @@ class UserModel(ModelMixin, UserMixin):
github_login: Mapped[str | None] = mapped_column(String(32), nullable=True, comment="Github登录")
wx_login: Mapped[str | None] = mapped_column(String(32), nullable=True, comment="微信登录")
qq_login: Mapped[str | None] = mapped_column(String(32), nullable=True, comment="QQ登录")
status: Mapped[int] = mapped_column(Integer, default=0, nullable=False, comment="状态(0:启动 1:停用)", index=True)
status: Mapped[int] = mapped_column(Integer, default=0, nullable=False, comment="状态(0:启动 1:停用)")
description: Mapped[str | None] = mapped_column(Text, default=None, nullable=True, comment="备注")
dept_id: Mapped[int | None] = mapped_column(Integer, ForeignKey("sys_dept.id", ondelete="SET NULL", onupdate="CASCADE"), nullable=True, index=True, comment="部门ID")
dept: Mapped["DeptModel | None"] = relationship(back_populates="users", foreign_keys=[dept_id], lazy="selectin")
roles: Mapped[list["RoleModel"]] = relationship(secondary="sys_user_roles", back_populates="users", lazy="selectin")
positions: Mapped[list["PositionModel"]] = relationship(secondary="sys_user_positions", back_populates="users", lazy="selectin")
created_by: Mapped["UserModel | None"] = relationship("UserModel", foreign_keys="UserModel.created_id", remote_side="UserModel.id", lazy="selectin", uselist=False, viewonly=True)
updated_by: Mapped["UserModel | None"] = relationship("UserModel", foreign_keys="UserModel.updated_id", remote_side="UserModel.id", lazy="selectin", uselist=False, viewonly=True)
deleted_by: Mapped["UserModel | None"] = relationship("UserModel", foreign_keys="UserModel.deleted_id", remote_side="UserModel.id", lazy="selectin", uselist=False, viewonly=True)
dept: Mapped["DeptModel | None"] = relationship(back_populates="users", foreign_keys=[dept_id])
roles: Mapped[list["RoleModel"]] = relationship(secondary="sys_user_roles", back_populates="users")
positions: Mapped[list["PositionModel"]] = relationship(secondary="sys_user_positions", back_populates="users")
created_by: Mapped["UserModel | None"] = relationship("UserModel", foreign_keys="UserModel.created_id", remote_side="UserModel.id", uselist=False, viewonly=True)
updated_by: Mapped["UserModel | None"] = relationship("UserModel", foreign_keys="UserModel.updated_id", remote_side="UserModel.id", uselist=False, viewonly=True)
deleted_by: Mapped["UserModel | None"] = relationship("UserModel", foreign_keys="UserModel.deleted_id", remote_side="UserModel.id", uselist=False, viewonly=True)
backend/app/api/v1/module_system/user/schema.py
+9 -10
View File
@@ -1,3 +1,4 @@
import re
from urllib.parse import urlparse
from pydantic import (
@@ -12,7 +13,7 @@ from pydantic import (
from app.api.v1.module_system.menu.schema import MenuTreeOutSchema
from app.api.v1.module_system.role.schema import RoleOutSchema
from app.core.base_schema import BaseQueryParam, BaseSchema, CommonSchema, CoreUserSchema, UserByQueryParam, UserBySchema
from app.core.validator import email_validator, mobile_validator
from app.core.validator import DateTimeStr, email_validator, mobile_validator
class CurrentUserUpdateSchema(BaseModel):
@@ -81,7 +82,6 @@ class UserForgetPasswordSchema(BaseModel):
v = value.strip()
if not v:
raise ValueError("账号不能为空")
import re
if not re.match(r"^[A-Za-z][A-Za-z0-9_.-]{2,31}$", v):
raise ValueError("账号需以字母开头，3-32 位，仅允许字母、数字、_ . -")
@@ -166,7 +166,6 @@ class UserCreateSchema(CurrentUserUpdateSchema):
if not value:
return value
v = value.strip()
import re
if not re.match(r"^[A-Za-z][A-Za-z0-9_.-]{1,31}$", v):
raise ValueError("账号需以字母开头，2-32 位，仅允许字母、数字、_ . -")
@@ -200,7 +199,6 @@ class UserRegisterSchema(BaseModel):
v = value.strip()
if not v:
raise ValueError("账号不能为空")
import re
if not re.match(r"^[A-Za-z][A-Za-z0-9_.-]{2,31}$", v):
raise ValueError("账号需以字母开头，3-32 位，仅允许字母、数字、_ . -")
@@ -245,7 +243,6 @@ class UserUpdateSchema(CurrentUserUpdateSchema):
if not value:
return value
v = value.strip()
import re
if not re.match(r"^[A-Za-z][A-Za-z0-9_.-]{1,31}$", v):
raise ValueError("账号需以字母开头，2-32 位，仅允许字母、数字、_ . -")
@@ -279,6 +276,7 @@ class UserOutSchema(CoreUserSchema, BaseSchema, UserBySchema):
roles: list[RoleOutSchema] | None = Field(default=[], description="角色")
menus: list[MenuTreeOutSchema] | None = Field(default=[], description="菜单")
is_superuser: bool = Field(default=False, description="是否超管")
last_login: DateTimeStr | None = Field(default=None, description="最后登录时间")
class UserQueryParam(BaseQueryParam, UserByQueryParam):
@@ -290,13 +288,14 @@ class UserQueryParam(BaseQueryParam, UserByQueryParam):
- 业务字段：用户名、名称、手机号、邮箱、部门、状态
"""
username: str | None = Field(None, description="用户名")
name: str | None = Field(None, description="名称")
mobile: str | None = Field(None, description="手机号", pattern=r"^1[3-9]\d{9}$")
username: str | None = Field(None, description="用户名", json_schema_extra={"q": "like"})
name: str | None = Field(None, description="名称", json_schema_extra={"q": "like"})
mobile: str | None = Field(None, description="手机号", pattern=r"^1[3-9]\d{9}$", json_schema_extra={"q": "eq"})
email: str | None = Field(
None,
description="邮箱",
pattern=r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$",
json_schema_extra={"q": "eq"},
)
dept_id: int | None = Field(None, description="部门ID")
status: int | None = Field(None, description="是否可用")
dept_id: int | None = Field(None, description="部门ID", json_schema_extra={"q": "eq"})
status: int | None = Field(None, description="是否可用", json_schema_extra={"q": "eq"})
backend/app/api/v1/module_system/user/service.py
+96 -132
View File
@@ -28,6 +28,8 @@ from .schema import (
UserUpdateSchema,
)
_USER_PRELOAD = ["dept", "positions", "roles.menus", "roles.depts"]
class UserService:
"""用户管理服务"""
@@ -37,11 +39,10 @@ class UserService:
self.db = db
async def detail(self, id: int) -> UserOutSchema:
user = await UserCRUD(self.auth, self.db).get_or_404(id=id)
user = await UserCRUD(self.auth, self.db).get_or_404(id=id, preload=_USER_PRELOAD)
result = UserOutSchema.model_validate(user)
if user.dept_id:
dept = await DeptCRUD(self.auth, self.db).get(id=user.dept_id)
result.dept_name = dept.name if dept else None
if user.dept:
result.dept_name = user.dept.name
return result
async def get_list(
@@ -49,8 +50,12 @@ class UserService:
search: UserQueryParam | None = None,
order_by: list[dict[str, str]] | None = None,
) -> list[UserOutSchema]:
user_list = await UserCRUD(self.auth, self.db).get_list(search=search_to_dict(search), order_by=order_by)
return [UserOutSchema.model_validate(user) for user in user_list]
user_list = await UserCRUD(self.auth, self.db).get_list(search=search_to_dict(search), order_by=order_by, preload=_USER_PRELOAD)
result = [UserOutSchema.model_validate(user) for user in user_list]
for user, item in zip(user_list, result, strict=True):
if user.dept:
item.dept_name = user.dept.name
return result
async def page(
self,
@@ -66,98 +71,71 @@ class UserService:
order_by=order_by or [{"id": "asc"}],
search=search_to_dict(search),
out_schema=UserOutSchema,
preload=_USER_PRELOAD,
)
async def create(self, data: UserCreateSchema) -> UserOutSchema:
if not data.username:
raise CustomException(msg="用户名不能为空")
if data.is_superuser:
raise CustomException(msg="不允许创建超级管理员")
user = await UserCRUD(self.auth, self.db).get(username=data.username)
if user:
if await UserCRUD(self.auth, self.db).get(username=data.username):
raise CustomException(msg="已存在相同用户名称的账号")
if data.dept_id and not await DeptCRUD(self.auth, self.db).get(id=data.dept_id):
raise CustomException(msg="该数据不存在")
if data.password:
data.password = PwdUtil.hash_password(password=data.password)
create_data = data.model_dump(exclude_none=True, exclude={"role_ids", "position_ids"})
new_user = await UserCRUD(self.auth, self.db).create(data=create_data)
if data.role_ids:
await UserCRUD(self.auth, self.db).set_user_roles(user_ids=[new_user.id], role_ids=data.role_ids)
if data.position_ids:
await UserCRUD(self.auth, self.db).set_user_positions(user_ids=[new_user.id], position_ids=data.position_ids)
return await self.detail(id=new_user.id)
async def update(self, id: int, data: UserUpdateSchema) -> UserOutSchema:
if data.username:
if exist_user := await UserCRUD(self.auth, self.db).get(username=data.username):
if exist_user.id != id:
raise CustomException(msg="更新失败，账号已存在")
if data.mobile:
if exist_mobile := await UserCRUD(self.auth, self.db).get(mobile=data.mobile):
if exist_mobile.id != id:
raise CustomException(msg="该数据已存在")
if data.email:
if exist_email := await UserCRUD(self.auth, self.db).get(email=data.email):
if exist_email.id != id:
raise CustomException(msg="该数据已存在")
if data.dept_id:
dept = await DeptCRUD(self.auth, self.db).get(id=data.dept_id)
if not dept:
raise CustomException(msg="该数据不存在")
if dept.status == 1:
raise CustomException(msg="部门已被禁用")
if data.password:
data.password = PwdUtil.hash_password(password=data.password)
new_user = await UserCRUD(self.auth, self.db).create(data=data)
if data.role_ids and len(data.role_ids) > 0:
await UserCRUD(self.auth, self.db).set_user_roles(user_ids=[new_user.id], role_ids=data.role_ids)
if data.position_ids and len(data.position_ids) > 0:
await UserCRUD(self.auth, self.db).set_user_positions(user_ids=[new_user.id], position_ids=data.position_ids)
return UserOutSchema.model_validate(new_user)
update_data = data.model_dump(exclude_unset=True, exclude_none=True, exclude={"role_ids", "position_ids"})
await UserCRUD(self.auth, self.db).update(id=id, data=update_data)
async def update(self, id: int, data: UserUpdateSchema) -> UserOutSchema:
if not data.username:
raise CustomException(msg="账号不能为空")
if data.role_ids:
roles = await RoleCRUD(self.auth, self.db).get_list(search={"id": ("in", data.role_ids)})
if len(roles) != len(data.role_ids):
raise CustomException(msg="更新失败，部分角色不存在")
if not all(role.status == 0 for role in roles):
raise CustomException(msg="更新失败，部分角色已被禁用")
await UserCRUD(self.auth, self.db).set_user_roles(user_ids=[id], role_ids=data.role_ids)
user = await UserCRUD(self.auth, self.db).get_or_404(id=id)
if user.is_superuser:
raise CustomException(msg="超级管理员不允许修改")
if data.position_ids:
positions = await PositionCRUD(self.auth, self.db).get_list(search={"id": ("in", data.position_ids)})
if len(positions) != len(data.position_ids):
raise CustomException(msg="更新失败，部分岗位不存在")
if not all(position.status == 0 for position in positions):
raise CustomException(msg="更新失败，部分岗位已被禁用")
await UserCRUD(self.auth, self.db).set_user_positions(user_ids=[id], position_ids=data.position_ids)
await self._validate_unique_username(data.username, exclude_id=id)
await self._validate_mobile_email_unique(data, exclude_id=id)
await self._validate_dept_active(data.dept_id)
new_user = await UserCRUD(self.auth, self.db).update(id=id, data=data)
await self._assign_user_roles(id, data.role_ids)
await self._assign_user_positions(id, data.position_ids)
return UserOutSchema.model_validate(new_user)
async def _validate_unique_username(self, username: str, exclude_id: int) -> None:
"""校验用户名唯一性"""
exist_user = await UserCRUD(self.auth, self.db).get(username=username)
if exist_user and exist_user.id != exclude_id:
raise CustomException(msg="更新失败，账号已存在")
async def _validate_mobile_email_unique(self, data: UserUpdateSchema, exclude_id: int) -> None:
"""校验手机号和邮箱唯一性"""
if data.mobile:
exist_mobile_user = await UserCRUD(self.auth, self.db).get(mobile=data.mobile)
if exist_mobile_user and exist_mobile_user.id != exclude_id:
raise CustomException(msg="该数据已存在")
if data.email:
exist_email_user = await UserCRUD(self.auth, self.db).get(email=data.email)
if exist_email_user and exist_email_user.id != exclude_id:
raise CustomException(msg="该数据已存在")
async def _validate_dept_active(self, dept_id: int | None) -> None:
"""校验部门存在且已启用"""
if not dept_id:
return
dept = await DeptCRUD(self.auth, self.db).get(id=dept_id)
if not dept:
raise CustomException(msg="该数据不存在")
if dept.status == 1:
raise CustomException(msg="部门已被禁用")
async def _assign_user_roles(self, user_id: int, role_ids: list[int] | None) -> None:
"""校验并分配用户角色"""
if not role_ids or len(role_ids) < 1:
return
roles = await RoleCRUD(self.auth, self.db).get_list(search={"id": ("in", role_ids)})
if len(roles) != len(role_ids):
raise CustomException(msg="更新失败，部分角色不存在")
if not all(role.status == 0 for role in roles):
raise CustomException(msg="更新失败，部分角色已被禁用")
await UserCRUD(self.auth, self.db).set_user_roles(user_ids=[user_id], role_ids=role_ids)
async def _assign_user_positions(self, user_id: int, position_ids: list[int] | None) -> None:
"""校验并分配用户岗位"""
if not position_ids or len(position_ids) < 1:
return
positions = await PositionCRUD(self.auth, self.db).get_list(search={"id": ("in", position_ids)})
if len(positions) != len(position_ids):
raise CustomException(msg="更新失败，部分岗位不存在")
if not all(position.status == 0 for position in positions):
raise CustomException(msg="更新失败，部分岗位已被禁用")
await UserCRUD(self.auth, self.db).set_user_positions(user_ids=[user_id], position_ids=position_ids)
return await self.detail(id=id)
async def delete(self, ids: list[int]) -> None:
if not ids:
@@ -173,25 +151,21 @@ class UserService:
if user.is_superuser:
errors.append(f"用户[{uid}]是超级管理员，不能删除")
continue
if user.status == 0:
errors.append(f"用户[{uid}]已启用，不能删除")
continue
if self.auth.user.id == uid:
errors.append("不能删除当前登陆用户")
continue
if errors:
raise CustomException(msg="; ".join(errors))
if not user_map:
raise CustomException(msg="删除对象不存在")
await UserCRUD(self.auth, self.db).set_user_roles(user_ids=ids, role_ids=[])
await UserCRUD(self.auth, self.db).set_user_positions(user_ids=ids, position_ids=[])
await UserCRUD(self.auth, self.db).delete(ids=ids)
async def current_info(self) -> UserOutSchema:
if not self.auth.user.id:
user_id = self.auth.user.id
if not user_id:
raise CustomException(msg="该数据不存在")
user = await UserCRUD(self.auth, self.db).get(id=self.auth.user.id)
user = await UserCRUD(self.auth, self.db).get(id=user_id, preload=_USER_PRELOAD)
if user is None:
raise CustomException(msg="该数据不存在")
user_dict = UserOutSchema.model_validate(user)
@@ -199,9 +173,9 @@ class UserService:
user_dict.dept_name = user.dept.name
user_dict.is_superuser = user.is_superuser
_pc_only = {"client": "pc"}
_pc_only = {"scope": "web"}
if self.auth.user.is_superuser:
menu_all = await MenuCRUD(self.auth, self.db).tree_list(
menu_all = await MenuCRUD(self.auth, self.db).get_list(
search={"type": ("in", [1, 2, 3, 4]), "status": 0, **_pc_only},
order_by=[{"order": "asc"}],
)
@@ -211,7 +185,7 @@ class UserService:
menus_raw = (
[
MenuOutSchema.model_validate(menu)
for menu in await MenuCRUD(self.auth, self.db).tree_list(
for menu in await MenuCRUD(self.auth, self.db).get_list(
search={"id": ("in", list(menu_ids)), **_pc_only},
order_by=[{"order": "asc"}],
)
@@ -225,24 +199,22 @@ class UserService:
return user_dict
async def update_current_info(self, data: CurrentUserUpdateSchema) -> UserOutSchema:
if not self.auth.user.id:
user_id = self.auth.user.id
if not user_id:
raise CustomException(msg="该数据不存在")
user = await UserCRUD(self.auth, self.db).get(id=self.auth.user.id)
if not user:
raise CustomException(msg="该数据不存在")
if user.is_superuser:
raise CustomException(msg="超级管理员不能修改个人信息")
if data.mobile:
exist_mobile_user = await UserCRUD(self.auth, self.db).get(mobile=data.mobile)
if exist_mobile_user and exist_mobile_user.id != self.auth.user.id:
raise CustomException(msg="该数据已存在")
if exist_mobile := await UserCRUD(self.auth, self.db).get(mobile=data.mobile):
if exist_mobile.id != user_id:
raise CustomException(msg="该数据已存在")
if data.email:
exist_email_user = await UserCRUD(self.auth, self.db).get(email=data.email)
if exist_email_user and exist_email_user.id != self.auth.user.id:
raise CustomException(msg="该数据已存在")
if exist_email := await UserCRUD(self.auth, self.db).get(email=data.email):
if exist_email.id != user_id:
raise CustomException(msg="该数据已存在")
user_update_data = UserUpdateSchema(**data.model_dump())
new_user = await UserCRUD(self.auth, self.db).update(id=self.auth.user.id, data=user_update_data)
return UserOutSchema.model_validate(new_user)
await UserCRUD(self.auth, self.db).update(id=user_id, data=user_update_data)
return await self.detail(id=user_id)
async def set_available(self, data: BatchSetAvailable) -> None:
users = await UserCRUD(self.auth, self.db).get_list(search={"id": ("in", data.ids)})
@@ -252,41 +224,29 @@ class UserService:
await UserCRUD(self.auth, self.db).set(ids=data.ids, status=data.status)
async def change_password(self, data: UserChangePasswordSchema) -> UserOutSchema:
if not self.auth.user.id:
user_id = self.auth.user.id
if not user_id:
raise CustomException(msg="该数据不存在")
if not data.old_password or not data.new_password:
raise CustomException(msg="密码不能为空")
user = await UserCRUD(self.auth, self.db).get(id=self.auth.user.id)
if not user:
raise CustomException(msg="该数据不存在")
user = await UserCRUD(self.auth, self.db).get_or_404(id=user_id)
if not PwdUtil.verify_password(plain_password=data.old_password, password_hash=user.password):
raise CustomException(msg="原密码输入错误")
new_password_hash = PwdUtil.hash_password(password=data.new_password)
new_user = await UserCRUD(self.auth, self.db).change_password(id=user.id, password_hash=new_password_hash)
return UserOutSchema.model_validate(new_user)
await UserCRUD(self.auth, self.db).change_password(id=user_id, password_hash=new_password_hash)
return await self.detail(id=user_id)
async def reset_password(self, data: ResetPasswordSchema) -> UserOutSchema:
if not data.password:
raise CustomException(msg="密码不能为空")
user = await UserCRUD(self.auth, self.db).get(id=data.id)
if not user:
raise CustomException(msg="该数据不存在")
user = await UserCRUD(self.auth, self.db).get_or_404(id=data.id)
if user.is_superuser:
raise CustomException(msg="超级管理员密码不能重置")
new_password_hash = PwdUtil.hash_password(password=data.password)
new_user = await UserCRUD(self.auth, self.db).change_password(id=data.id, password_hash=new_password_hash)
return UserOutSchema.model_validate(new_user)
await UserCRUD(self.auth, self.db).change_password(id=data.id, password_hash=new_password_hash)
return await self.detail(id=data.id)
async def forget_password(self, data: UserForgetPasswordSchema) -> UserOutSchema:
# 直接按用户名查找用户
user = await UserCRUD(self.auth, self.db).get(username=data.username)
if not user:
raise CustomException(msg="该数据不存在")
user = await UserCRUD(self.auth, self.db).get_or_404(username=data.username)
if user.status == 1:
raise CustomException(msg="用户已停用")
if user.is_superuser:
@@ -295,8 +255,8 @@ class UserService:
raise CustomException(msg="手机号不匹配")
new_password_hash = PwdUtil.hash_password(password=data.new_password)
new_user = await UserCRUD(self.auth, self.db).forget_password(id=user.id, password_hash=new_password_hash)
return UserOutSchema.model_validate(new_user)
await UserCRUD(self.auth, self.db).change_password(id=user.id, password_hash=new_password_hash)
return await self.detail(id=user.id)
async def register(self, data: UserRegisterSchema) -> UserOutSchema:
"""用户注册"""
@@ -312,8 +272,10 @@ class UserService:
status=0,
)
new_user = await UserCRUD(self.auth, self.db).create_obj_crud(data=create_data)
if not new_user:
raise CustomException(msg="注册失败")
logger.info(f"新用户注册成功: {data.username}")
return UserOutSchema.model_validate(new_user)
return await self.detail(id=new_user.id)
async def batch_import(self, file: UploadFile, update_support: bool = False) -> str:
header_dict = {
@@ -426,7 +388,9 @@ class UserService:
return 0, f"第{row_num}行: 用户 {user_data['username']} 已存在"
else:
user_create_schema = UserCreateSchema(**user_data)
new_user = await UserCRUD(self.auth, self.db).create(data=user_create_schema)
new_user = await UserCRUD(self.auth, self.db).create(
data=user_create_schema.model_dump(exclude_none=True, exclude={"role_ids", "position_ids"}) # type: ignore[arg-type]
)
if user_create_schema.role_ids and len(user_create_schema.role_ids) > 0:
await UserCRUD(self.auth, self.db).set_user_roles(
user_ids=[new_user.id], role_ids=user_create_schema.role_ids
backend/app/api/v1/module_system/versions/controller.py
+2 -2
View File
@@ -1,6 +1,6 @@
from typing import Annotated
from fastapi import APIRouter, Body, Depends, Path, Query, Security, status
from fastapi import APIRouter, Body, Depends, Path, Security, status
from fastapi.responses import JSONResponse
from sqlalchemy.ext.asyncio import AsyncSession
@@ -26,7 +26,7 @@ async def get_version_list_controller(
auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:version:query"]))],
db: Annotated[AsyncSession, Depends(db_getter)],
page: Annotated[PaginationQueryParam, Depends()],
search: Annotated[VersionQueryParam, Query()],
search: Annotated[VersionQueryParam, Depends()],
) -> JSONResponse:
service = VersionService(auth, db)
result = await service.page(page_no=page.page_no, page_size=page.page_size, search=search)
backend/app/api/v1/module_system/versions/model.py
+1 -2
View File
@@ -9,9 +9,8 @@ class VersionModel(ModelMixin, UserMixin):
__tablename__: str = "sys_version"
__table_args__: dict[str, str] = {"comment": "版本管理表"}
__loader_options__: list[str] = ["created_by", "updated_by", "deleted_by"]
version: Mapped[str] = mapped_column(String(50), nullable=False, comment="版本号")
version: Mapped[str] = mapped_column(String(32), nullable=False, unique=True, comment="版本号")
title: Mapped[str] = mapped_column(String(200), nullable=False, comment="版本标题")
date: Mapped[str] = mapped_column(String(50), nullable=False, comment="发布日期")
content: Mapped[str | None] = mapped_column(Text, nullable=True, default=None, comment="版本富文本内容")
backend/app/api/v1/module_system/versions/schema.py
+1 -1
View File
@@ -47,4 +47,4 @@ class VersionStatusSchema(BaseModel):
class VersionQueryParam(BaseQueryParam):
"""版本查询参数"""
status: int | None = Field(default=None, description="状态: 0=草稿,1=已发布,2=已回滚")
status: int | None = Field(default=None, description="状态: 0=草稿,1=已发布,2=已回滚", json_schema_extra={"q": "eq"})