djangoadmin/DjangoAdmin_FastAPI_EleVue

Fork 0

Files

T

gitadmin 3233d221b0 签入版本

2026-09-12 17:39:13 +08:00

17 KiB

Raw Blame History

系统架构说明

本文档详细说明 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 / MSSQL DBCC 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() 原始出口供复杂查询使用

📚 相关文档

部署指南 — Docker / 手动 / 生产环境部署
数据库迁移 — 跨库迁移操作手册
开发文档 — 在线开发文档

17 KiB Raw Blame History Unescape Escape