17 KiB
17 KiB
系统架构说明
本文档详细说明 DjangoAdmin FastAPI + EleVue 版本的整体架构设计、分层模式、核心模块以及关键设计决策。
📐 整体架构
前后端分离架构
┌─────────────────────────────────────────────────────────────┐
│ 前端 (Vue 3) │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Vue 3 │ │ Element │ │ Pinia │ │ Vue │ │
│ │ + TS │ │ Plus │ │ 状态管理 │ │ Router │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
│ ↕ Axios HTTP Client (Bearer Token) │
└─────────────────────────────────────────────────────────────┘
↕ JSON / JWT
┌─────────────────────────────────────────────────────────────┐
│ 后端 (FastAPI) │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 中间件链 │ │
│ │ CORS → Redis注入 → 限流 → DB会话 → 操作日志 → 日志 │ │
│ └─────────────────────────────────────────────────────┘ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ API 路由层 │ │
│ │ /api/v1/auth/* (免认证) /api/v1/* (login_required) │ │
│ │ @permission_required @check_demo │ │
│ └─────────────────────────────────────────────────────┘ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 业务模块层 │ │
│ │ Service → Repository → Model │ │
│ └─────────────────────────────────────────────────────┘ │
│ ↕ ↕ │
│ ┌──────────┐ ┌──────────┐ │
│ │ MySQL / │ │ Redis │ │
│ │ PG / │ │ 缓存 │ │
│ │ MSSQL / │ │ 会话 │ │
│ │ Oracle │ │ 黑名单 │ │
│ └──────────┘ └──────────┘ │
└─────────────────────────────────────────────────────────────┘
🔄 请求处理流程
完整请求生命周期
1. 客户端发起 HTTP 请求
↓
2. CORS 中间件（src/middleware/cors.py）
- 处理跨域预检请求
↓
3. Redis 注入中间件（src/core/redis_client.py: redis_middleware）
- 将异步 Redis 客户端注入 contextvars
↓
4. 限流中间件（src/middleware/rate_limit.py）
- 基于 Redis 滑动窗口的 per-IP 限流
- 超限返回 {code: 1} 标准响应
↓
5. DB 会话中间件（src/middleware/db_session.py）
- 创建请求级数据库会话
- 请求结束后自动 commit/rollback
↓
6. 操作日志中间件（src/middleware/operation_log.py）
- 记录写操作（POST/PUT/DELETE）的行为日志
↓
7. 路由匹配
- auth 路由（/api/v1/auth/*）：免认证
- v1 路由（/api/v1/*）：login_required 依赖校验 JWT
↓
8. 权限装饰器 @permission_required
- 校验当前用户是否有该接口的权限
- 用户 ID 1（admin）自动跳过权限检查
↓
9. 演示模式装饰器 @check_demo
- 演示环境下拦截写操作
↓
10. 端点处理函数
- 参数由 Pydantic 自动校验
- 调用 Service 层业务逻辑
- 返回 R.ok() / R.failed() 统一响应
↓
11. 异常处理
- RequestValidationError → {code: 1, msg: "field: message"}
- BusinessException → {code: 1, msg: "业务错误"}
- AuthorizationException → {code: 401, msg: "认证失败"}
📦 分层模块模式
四层架构
每个业务模块遵循统一的四层架构设计：
src/modules/{group}/{name}/
├── models.py # 数据模型层 — SQLAlchemy ORM 模型定义
├── schemas.py # 数据校验层 — Pydantic 请求/响应模型
├── repository.py # 数据访问层 — 通用 CRUD 与自定义查询
└── service.py # 业务逻辑层 — 事务管理、缓存、业务规则
Model 层（数据模型）
# src/modules/system/level/models.py
from core.base_model import base_model
from core.base_db import base_db
from core.config.database import DB_PREFIX
class SysLevel(base_model, base_db):
"""职级表"""
__tablename__ = DB_PREFIX + "level"
name = Column(String(100), nullable=False, comment="职级名称")
code = Column(String(50), nullable=False, comment="职级编码")
sort = Column(Integer, default=0, comment="排序")
status = Column(Integer, default=1, comment="状态 1=启用 0=禁用")
remark = Column(String(500), comment="备注")
- 继承
base_model：提供id,create_user,create_time,update_user,update_time,is_delete - 继承
base_db：提供save(),delete()实例方法 - 表名使用
DB_PREFIX前缀，支持多项目共库
Schema 层（数据校验）
# src/modules/system/level/schemas.py
from pydantic import BaseModel, Field, field_validator
class LevelForm(BaseModel):
"""职级表单"""
id: Optional[int] = Field(None, description="ID")
name: str = Field(..., min_length=1, max_length=100, description="职级名称")
code: str = Field(..., min_length=1, max_length=50, description="职级编码")
sort: int = Field(0, ge=0, description="排序")
status: int = Field(1, description="状态")
remark: Optional[str] = Field(None, max_length=500, description="备注")
@field_validator('name')
@classmethod
def validate_name(cls, v):
if not v.strip():
raise ValueError("职级名称不能为空")
return v.strip()
- 继承 Pydantic
BaseModel - 使用
Field(...)定义约束（min_length,max_length,ge,le） - 使用
@field_validator自定义校验逻辑
Repository 层（数据访问）
# src/modules/system/level/repository.py
from core.base_repository import BaseRepository
from .models import SysLevel
class LevelRepository(BaseRepository[SysLevel]):
model = SysLevel
level_repo = LevelRepository()
- 继承
BaseRepository[Model]，自动获得通用 CRUD 方法 - 模块级单例
{name}_repo，全局复用 - 基类提供：
get_by_id,get_one,get_all,exists,exists_by_field,count,filter,filter_by,create,update,paginate,batch_delete - 等值查询自动软删过滤且跳过 None 值
Service 层（业务逻辑）
# src/modules/system/level/service.py
from core.base_service import BaseService
from .models import SysLevel
from .repository import level_repo
class LevelService(BaseService[SysLevel]):
repo = level_repo
model = SysLevel
page_like_fields = ["name", "code"]
page_eq_fields = ["status"]
page_order_by = ["sort", "id"]
unique_fields = [("name", "职级名称"), ("code", "职级编码")]
level_service = LevelService()
- 继承
BaseService[Model]，声明差异即可获得完整 CRUD - 可覆盖钩子：
_apply_page_filters,_serialize,_build_create_fields,_build_update_fields - 删除统一使用
BaseService.batch_delete_with_r(repo, ids)静态方法
HTTP 层（API 端点）
# src/api/v1/endpoints/level.py
from fastapi import APIRouter, Request
from core.access_decorators import permission_required, check_demo
router = APIRouter()
@router.get('/list')
@permission_required("sys:level:list")
async def list(request: Request):
"""职级列表"""
return await level_service.page_list(request)
@router.post('/add')
@permission_required("sys:level:add")
@check_demo
async def add(request: Request, data: LevelForm):
"""新增职级"""
return await level_service.add(request, data)
- 路由定义、权限装饰器、请求/响应处理
- 无业务逻辑，纯委托给 Service 层
🔐 认证与授权
JWT 认证流程
1. 登录（/api/v1/auth/login）
- 验证用户名 + 密码（bcrypt）
- 验证图形验证码
- 生成 JWT Token（HS256 签名）
- 返回 {access_token, token_type}
2. 请求认证（login_required 依赖）
- 从 Authorization: Bearer <token> 提取 Token
- 解析 JWT，验证签名与过期时间
- 检查 Token 是否在 Redis 黑名单中
- 设置 request.state.user_id / username / realname
3. Token 刷新（/api/v1/auth/refresh）
- 使用当前 Token 获取新 Token
- 旧 Token 加入 Redis 黑名单
4. 登出（/api/v1/auth/logout）
- 当前 Token 加入 Redis 黑名单
RBAC 权限模型
用户(User) ←→ 角色(Role) ←→ 菜单(Menu)
│ │ │
└─ user_role └─ role_menu └─ 权限标识
(sys:level:list)
- 用户-角色：多对多关系（
sys_user_role中间表） - 角色-菜单：多对多关系（
sys_role_menu中间表） - 菜单：存储权限标识（如
sys:level:list），支持目录/菜单/按钮三种类型 - 权限检查：
@permission_required("sys:level:list")装饰器在端点执行前校验 - 管理员豁免：用户 ID 1（admin）自动跳过所有权限检查
🗄️ 数据库设计
多数据库支持
框架通过 DB_DRIVER 环境变量切换数据库方言：
# src/core/config/database.py
DB_DRIVER = os.getenv("DB_DRIVER", "mysql")
# 驱动注册表
DRIVERS = {
"mysql": "pymysql",
"postgresql": "psycopg",
"mssql": "pymssql",
"oracle": "oracledb",
"sqlite": "sqlite3",
}
- SQLAlchemy 引擎根据驱动自动选择方言
- Model 层使用 SQLAlchemy 声明式映射，与具体数据库解耦
- Alembic 迁移脚本自动适配方言 DDL
软删除机制
所有业务表使用 is_delete 字段实现软删除：
is_delete = 0：正常记录is_delete = 1：已删除记录
Repository 基类的等值查询方法（get_one, get_all, count, filter_by, exists_by_field）自动过滤已删除记录。
跨库数据迁移
scripts/migrate_db.py 支持从 MySQL 迁移数据到任意目标数据库：
- 拓扑排序确定表依赖关系
- 分块
SELECT * ... ORDER BY id+executemany显式带 ID 插入 - 自增序列重置（PG
setval/ MSSQLDBCC CHECKIDENT） - 逐表
COUNT(*)校验数据一致性
⚡ 同步/异步约定
| 场景 | 方式 | 说明 |
|---|---|---|
| 纯 DB CRUD | 同步 def |
FastAPI 自动在线程池执行，不阻塞事件循环 |
需要 await（Redis/异步IO） |
async def + asyncio.to_thread |
同步操作用 to_thread 包裹 |
| 批量删除端点 | async def |
需 await parse_batch_ids(request) 异步读请求体 |
禁止在 async def 端点或 Service 内直接调用同步 DB 查询（会阻塞事件循环）。
📁 目录结构
DjangoAdmin_FastAPI_EleVue/
├── src/ # 后端源码
│ ├── main.py # 应用入口
│ ├── core/ # 核心基础设施
│ │ ├── app.py # 应用工厂（create_app）
│ │ ├── config/ # 配置模块（app/database/redis/auth/captcha/email/sms）
│ │ ├── database.py # 引擎/会话/上下文
│ │ ├── redis_client.py # Redis 异步客户端
│ │ ├── jwt.py # JWT 认证
│ │ ├── password.py # 密码加密
│ │ ├── response.py # 统一响应封装
│ │ ├── exceptions.py # 异常定义
│ │ ├── security.py # 安全工具
│ │ ├── access_decorators.py # 权限/演示模式装饰器
│ │ ├── logger.py # 日志配置
│ │ ├── base_model.py # ORM 基础模型
│ │ ├── base_db.py # 数据库操作基类
│ │ ├── base_repository.py # Repository 基类
│ │ └── base_service.py # Service 基类
│ ├── api/ # API 路由层
│ │ ├── __init__.py # 路由注册（register_router）
│ │ ├── deps.py # login_required 依赖
│ │ └── v1/
│ │ ├── router.py # 路由汇总
│ │ └── endpoints/ # 各模块端点定义
│ ├── modules/ # 业务模块层
│ │ ├── auth/ # 认证模块
│ │ ├── system/ # 系统管理（user/role/menu/dept/position/level）
│ │ ├── dictionary/ # 字典管理
│ │ ├── configuration/ # 配置管理
│ │ ├── article/ # 文章管理
│ │ ├── notice/ # 通知公告
│ │ ├── job/ # 定时任务
│ │ └── ... # 其他业务模块
│ ├── middleware/ # HTTP 中间件
│ │ ├── cors.py # 跨域处理
│ │ ├── rate_limit.py # 限流中间件
│ │ ├── db_session.py # DB 会话管理
│ │ ├── operation_log.py # 操作日志
│ │ └── logger.py # 请求日志
│ ├── constants/ # 常量与枚举
│ └── utils/ # 工具函数
├── ui/ # 前端源码（Vue 3 + TypeScript）
├── scripts/ # 运维脚本
│ ├── init_db.py # 数据库初始化
│ ├── seed_dict.py # 字典数据种子
│ ├── migrate_db.py # 跨库数据迁移
│ └── demo_job_loop.py # 定时任务演示
├── tests/ # 测试代码
│ ├── contract/ # 契约快照测试
│ ├── unit/ # 单元测试
│ └── integration/ # 集成测试
├── migrations/ # Alembic 迁移脚本
├── document/ # 项目文档
├── .env # 环境变量配置
├── docker-compose.yml # Docker 编排
├── Dockerfile # Docker 镜像构建
├── Makefile # 常用命令
├── requirements.txt # Python 依赖
├── alembic.ini # Alembic 配置
└── pytest.ini # 测试配置
🔧 核心设计决策
1. 为什么选择 Repository + Service 双层？
- Repository 封装数据访问细节，提供通用 CRUD，隔离 ORM 实现
- Service 封装业务逻辑，管理事务、缓存、权限等横切关注点
- 两层分离使得业务逻辑可独立测试，数据访问可独立优化
2. 为什么使用 contextvars 管理 DB 会话？
- FastAPI 的依赖注入不适合跨层级传递数据库会话
contextvars提供协程安全的请求级上下文- 中间件创建会话，
get_db_session()在任意层级获取当前会话 - 请求结束自动 commit/rollback，无需手动管理
3. 为什么使用软删除？
- 保留历史数据，支持数据审计和恢复
- 避免外键级联删除导致数据丢失
- Repository 基类自动过滤，业务层无需关心
4. 为什么 BaseRepository 等值查询跳过 None？
- 避免
filter_by(name=None)生成WHERE name = NULL（SQL 中应为IS NULL） - 简化 Service 层的条件构建，无需逐一判断 None
- 提供
filter()原始出口供复杂查询使用