559 lines
18 KiB
Markdown
559 lines
18 KiB
Markdown
# 扩展性设计
|
||
|
||
本章详细描述系统的扩展性设计，包括模块可插拔、数据库驱动可切换、中间件可插拔三大扩展维度。
|
||
|
||
:::tip 扩展性设计原则
|
||
遵循开闭原则（OCP）：对扩展开放，对修改关闭。新功能通过添加新模块实现，而非修改现有代码。核心框架通过基类、装饰器、中间件等机制提供扩展点，业务模块只需声明差异即可接入。
|
||
:::
|
||
|
||
## 一、模块可插拔设计
|
||
|
||
### 模块标准结构
|
||
|
||
每个业务模块遵循固定的四文件结构，新增模块只需创建这四个文件即可接入系统：
|
||
|
||
```
|
||
src/modules/{group}/{name}/
|
||
├── models.py # ORM 模型
|
||
├── repository.py # 数据访问层
|
||
├── schemas.py # 请求校验
|
||
└── service.py # 业务逻辑层
|
||
```
|
||
|
||
### 新增模块步骤
|
||
|
||
以新增「培训管理（Training）」模块为例。
|
||
|
||
**步骤一：创建模块目录和文件**
|
||
|
||
创建 `src/modules/system/training/` 目录，包含以下四个文件：
|
||
|
||
`models.py` — ORM 模型：
|
||
|
||
```python
|
||
# src/modules/system/training/models.py
|
||
from sqlalchemy import Column, String, Integer, text
|
||
from core.base_db import base_db
|
||
from core.base_model import base_model
|
||
from core.config import DB_PREFIX
|
||
|
||
|
||
class Training(base_model, base_db):
|
||
__tablename__ = DB_PREFIX + "training"
|
||
__table_comment__ = "培训记录表"
|
||
|
||
title = Column(String(200), nullable=False, comment="培训标题")
|
||
content = Column(String(2000), nullable=True, comment="培训内容")
|
||
trainer = Column(String(50), nullable=True, comment="培训讲师")
|
||
status = Column(Integer, default=1, server_default=text('1'), comment="状态")
|
||
sort = Column(Integer, default=0, server_default=text('0'), comment="排序")
|
||
```
|
||
|
||
`repository.py` — 数据访问层：
|
||
|
||
```python
|
||
# src/modules/system/training/repository.py
|
||
from modules.system.training.models import Training
|
||
from core.base_repository import BaseRepository
|
||
|
||
|
||
class TrainingRepository(BaseRepository[Training]):
|
||
pass
|
||
|
||
|
||
training_repo = TrainingRepository(Training)
|
||
```
|
||
|
||
`schemas.py` — 请求校验：
|
||
|
||
```python
|
||
# src/modules/system/training/schemas.py
|
||
from pydantic import BaseModel, Field
|
||
from core.base_schemas import BaseForm
|
||
|
||
|
||
class TrainingForm(BaseForm):
|
||
title: str = Field(..., min_length=1, max_length=200, description="培训标题")
|
||
content: str = Field(None, max_length=2000, description="培训内容")
|
||
trainer: str = Field(None, max_length=50, description="培训讲师")
|
||
status: int = Field(..., ge=1, le=2, description="状态")
|
||
sort: int = Field(..., ge=0, le=99999, description="排序")
|
||
|
||
|
||
class TrainingStatusForm(BaseModel):
|
||
id: int = Field(..., gt=0, description="培训ID")
|
||
status: int = Field(..., ge=1, le=2, description="状态")
|
||
```
|
||
|
||
`service.py` — 业务逻辑层：
|
||
|
||
```python
|
||
# src/modules/system/training/service.py
|
||
from core.base_service import BaseService
|
||
from modules.system.training.models import Training
|
||
from modules.system.training.repository import training_repo
|
||
|
||
|
||
class TrainingService(BaseService[Training]):
|
||
repo = training_repo
|
||
model = Training
|
||
page_like_fields = ('title', 'trainer')
|
||
page_eq_fields = ('status',)
|
||
page_order_by = (('sort', 'asc'),)
|
||
unique_fields = {'title': '培训标题不能重复'}
|
||
|
||
|
||
training_service = TrainingService()
|
||
```
|
||
|
||
**步骤二：创建 HTTP 端点**
|
||
|
||
创建 `src/api/v1/endpoints/training.py`：
|
||
|
||
```python
|
||
# src/api/v1/endpoints/training.py
|
||
from fastapi import APIRouter, Request
|
||
from modules.system.training.schemas import TrainingForm, TrainingStatusForm
|
||
from core.access_decorators import permission_required, check_demo
|
||
from modules.system.training.service import training_service
|
||
from core import R
|
||
|
||
router = APIRouter()
|
||
|
||
|
||
@router.get('/page')
|
||
@permission_required("sys:training:page")
|
||
def page(request: Request):
|
||
return training_service.get_page(request)
|
||
|
||
|
||
@router.get('/detail/{id}')
|
||
@permission_required("sys:training:detail")
|
||
def detail(request: Request, id):
|
||
return R.ok(data=training_service.get_detail(id))
|
||
|
||
|
||
@router.post('/add')
|
||
@permission_required("sys:training:add")
|
||
@check_demo
|
||
def add(request: Request, data: TrainingForm):
|
||
return training_service.add(request, data)
|
||
|
||
|
||
@router.put('/update')
|
||
@permission_required("sys:training:update")
|
||
@check_demo
|
||
def update(request: Request, data: TrainingForm):
|
||
return training_service.update(request, data)
|
||
|
||
|
||
@router.delete('/delete/{id}')
|
||
@permission_required("sys:training:delete")
|
||
@check_demo
|
||
def delete(request: Request, id):
|
||
return training_service.delete(id)
|
||
|
||
|
||
@router.put('/status')
|
||
@permission_required("sys:training:status")
|
||
@check_demo
|
||
def status(request: Request, data: TrainingStatusForm):
|
||
return training_service.update_status(request, data)
|
||
|
||
|
||
@router.delete('/batchDelete')
|
||
@permission_required("sys:training:batchDelete")
|
||
@check_demo
|
||
async def batch_delete(request: Request):
|
||
return await training_service.batch_delete(request)
|
||
```
|
||
|
||
**步骤三：注册路由**
|
||
|
||
在 `src/api/v1/router.py` 中添加：
|
||
|
||
```python
|
||
# src/api/v1/router.py
|
||
from api.v1.endpoints.training import router as training_router
|
||
|
||
v1.include_router(training_router, prefix="/training", tags=["培训管理"])
|
||
```
|
||
|
||
路由注册后，该模块的所有端点自动挂在 `/api/v1/training` 前缀下，并通过 `src/api/__init__.py` 的 `register_router()` 统一挂载到 FastAPI 应用实例。v1 路由器自带 `login_required` 依赖，无需额外配置认证。
|
||
|
||
**步骤四：配置菜单权限**
|
||
|
||
在数据库 `fastapi_menu` 表中插入对应的菜单和权限节点记录。权限标识格式为 `sys:{module}:{action}`，如 `sys:training:add`、`sys:training:page`。
|
||
|
||
### 模块解耦
|
||
|
||
模块之间通过以下方式解耦：
|
||
|
||
| 解耦方式 | 说明 | 示例 |
|
||
|:---|:---|:---|
|
||
| **Repository 单例** | 模块级单例，通过 import 引用 | `from modules.system.user.repository import user_repo` |
|
||
| **Service 单例** | 模块级单例，通过 import 引用 | `from modules.system.user.service import user_service` |
|
||
| **跨模块引用** | 通过 Repository 引用其他模块的模型 | PositionService 引用 User 模型检查引用 |
|
||
| **事件机制** | 暂未实现（未来可通过事件总线解耦） | — |
|
||
|
||
### 复杂模块扩展
|
||
|
||
对于业务逻辑复杂的模块（如 User、Role、Menu），在继承 `BaseService` 的基础上添加自定义方法：
|
||
|
||
```python
|
||
# src/modules/system/user/service.py
|
||
class UserService(BaseService[User]):
|
||
repo = user_repo
|
||
model = User
|
||
page_like_fields = ('username', 'realname')
|
||
page_eq_fields = ('status', 'dept_id')
|
||
|
||
# 自定义方法：用户详情（含角色列表）
|
||
def get_user_detail(self, user_id):
|
||
user = self.repo.get_by_id(user_id)
|
||
if not user:
|
||
return None
|
||
data = user.to_dict()
|
||
data['roleIds'] = get_user_role_ids(user_id)
|
||
data['roleNames'] = get_user_role_names(user_id)
|
||
return data
|
||
|
||
# 自定义方法：修改密码
|
||
def change_password(self, user_id, old_pwd, new_pwd):
|
||
user = self.repo.get_by_id(user_id)
|
||
if not verify_password(old_pwd, user.password):
|
||
return R.failed("原密码错误")
|
||
user.password = hash_password(new_pwd)
|
||
user.save()
|
||
return R.ok(msg="密码修改成功")
|
||
|
||
# 覆盖钩子：新增前哈希密码
|
||
def _before_add(self, request, data):
|
||
data.password = hash_password(data.password or DEFAULT_PASSWORD)
|
||
|
||
# 覆盖钩子：删除前检查
|
||
def _before_delete(self, ids):
|
||
if 1 in parse_id_list(str(ids)):
|
||
return "超级管理员不能删除"
|
||
return None
|
||
```
|
||
|
||
## 二、数据库驱动可切换
|
||
|
||
### 驱动注册机制
|
||
|
||
项目通过 `_SUPPORTED_DRIVERS` 声明支持的驱动，`_normalize_driver()` 处理别名映射和校验，`build_database_url()` 按驱动构建连接串：
|
||
|
||
```python
|
||
# src/core/config/database.py（简化）
|
||
import os
|
||
from urllib.parse import quote_plus
|
||
|
||
# 支持的数据库驱动
|
||
_SUPPORTED_DRIVERS = ("mysql", "postgresql", "mssql", "sqlite", "oracle")
|
||
|
||
|
||
def _normalize_driver(driver: str) -> str:
|
||
"""归一化 DB_DRIVER：别名映射 + 校验"""
|
||
name = str(driver).strip().lower()
|
||
alias_map = {
|
||
'postgres': 'postgresql',
|
||
'pg': 'postgresql',
|
||
'sqlserver': 'mssql',
|
||
'sql_server': 'mssql',
|
||
'sqlite3': 'sqlite',
|
||
}
|
||
name = alias_map.get(name, name)
|
||
if name not in _SUPPORTED_DRIVERS:
|
||
raise ValueError(f"不支持的 DB_DRIVER: {driver!r}")
|
||
return name
|
||
|
||
|
||
DB_DRIVER = _normalize_driver(os.getenv('DB_DRIVER', 'mysql'))
|
||
DB_HOST = os.getenv('DB_HOST', '127.0.0.1')
|
||
DB_PORT = int(os.getenv('DB_PORT', '3306'))
|
||
DB_DATABASE = os.getenv('DB_DATABASE', 'djangoadmin.fastapi.elevue')
|
||
DB_USERNAME = os.getenv('DB_USERNAME', 'root')
|
||
DB_PASSWORD = os.getenv('DB_PASSWORD', '')
|
||
|
||
|
||
def build_database_url(driver: str = None) -> str:
|
||
"""按驱动构建 SQLAlchemy 连接串"""
|
||
drv = _normalize_driver(driver if driver is not None else DB_DRIVER)
|
||
|
||
if drv == 'sqlite':
|
||
return 'sqlite:///./' + DB_DATABASE
|
||
|
||
base = quote_plus(DB_USERNAME) + ':' + quote_plus(DB_PASSWORD) + '@' + DB_HOST + ':' + str(DB_PORT)
|
||
|
||
if drv == 'mysql':
|
||
return 'mysql+pymysql://' + base + '/' + DB_DATABASE + '?charset=utf8mb4'
|
||
if drv == 'postgresql':
|
||
return 'postgresql+psycopg://' + base + '/' + DB_DATABASE
|
||
if drv == 'oracle':
|
||
return 'oracle+oracledb://' + base + '/?service_name=' + quote_plus(DB_DATABASE)
|
||
# mssql
|
||
return 'mssql+pymssql://' + base + '/' + DB_DATABASE + '?charset=utf8'
|
||
```
|
||
|
||
各驱动的连接串格式：
|
||
|
||
| 驱动 | 连接串格式 | Python 包 |
|
||
|:----|:----|:----|
|
||
| mysql | `mysql+pymysql://user:pass@host:port/db?charset=utf8mb4` | PyMySQL |
|
||
| postgresql | `postgresql+psycopg://user:pass@host:port/db` | psycopg[binary] |
|
||
| mssql | `mssql+pymssql://user:pass@host:port/db?charset=utf8` | pymssql |
|
||
| oracle | `oracle+oracledb://user:pass@host:port/?service_name=db` | oracledb |
|
||
| sqlite | `sqlite:///./path/to/db` | 内置 |
|
||
|
||
### 切换方式
|
||
|
||
只需修改 `.env` 文件中的 `DB_DRIVER` 和对应的连接参数：
|
||
|
||
```bash
|
||
# MySQL（默认）
|
||
DB_DRIVER=mysql
|
||
DB_HOST=127.0.0.1
|
||
DB_PORT=3306
|
||
DB_NAME=djangoadmin
|
||
DB_USER=root
|
||
DB_PASSWORD=your_password
|
||
|
||
# PostgreSQL
|
||
DB_DRIVER=postgresql
|
||
DB_HOST=127.0.0.1
|
||
DB_PORT=5432
|
||
DB_NAME=djangoadmin
|
||
DB_USER=postgres
|
||
DB_PASSWORD=your_password
|
||
|
||
# SQLite
|
||
DB_DRIVER=sqlite
|
||
DB_DATABASE=./data.db
|
||
|
||
# SQL Server
|
||
DB_DRIVER=mssql
|
||
DB_HOST=127.0.0.1
|
||
DB_PORT=1433
|
||
DB_NAME=djangoadmin
|
||
DB_USER=sa
|
||
DB_PASSWORD=your_password
|
||
```
|
||
|
||
:::warning 驱动切换注意事项
|
||
1. 切换驱动后需安装对应的 Python 包（如 `psycopg[binary]` for PostgreSQL）
|
||
2. 数据迁移使用 `scripts/migrate_db.py` 工具
|
||
3. SQLite 不支持并发写入，仅适用于开发/测试环境
|
||
4. Oracle 需要额外的客户端库配置
|
||
:::
|
||
|
||
### 跨库数据迁移
|
||
|
||
```bash
|
||
# 干跑：核对迁移计划
|
||
python scripts/migrate_db.py --src-driver mysql --src-host ... --dst-driver postgresql --dst-host ... --dry-run
|
||
|
||
# 执行迁移
|
||
python scripts/migrate_db.py --src-driver mysql --src-host ... --dst-driver postgresql --dst-host ...
|
||
|
||
# 幂等重跑（先删除目标库数据）
|
||
python scripts/migrate_db.py --src-driver mysql --src-host ... --dst-driver postgresql --dst-host ... --drop-target-first
|
||
```
|
||
|
||
## 三、中间件可插拔
|
||
|
||
### 中间件注册机制
|
||
|
||
所有中间件在 `src/core/app.py` 的 `create_app()` 中按顺序注册。新增中间件只需：
|
||
|
||
1. 在 `src/middleware/` 目录创建中间件文件
|
||
2. 在 `create_app()` 中注册
|
||
|
||
```python
|
||
# src/core/app.py - create_app() 中间件注册顺序
|
||
def create_app() -> FastAPI:
|
||
app = FastAPI(title="FastAPI", version="v1", lifespan=_combined_lifespan)
|
||
|
||
# 先注册 → 内层（最后处理请求）
|
||
app.middleware("http")(redis_middleware) # Redis 上下文注入
|
||
app.middleware("http")(db_session_middleware) # DB 会话管理（commit/rollback）
|
||
app.middleware("http")(operation_log_middleware) # 操作日志记录
|
||
app.middleware("http")(token_refresh_middleware) # Token 续签（X-Refresh-Token 响应头）
|
||
register_router(app) # 路由注册（login_required 依赖）
|
||
register_middleware(app) # 登录检测钩子
|
||
register_exception(app) # 全局异常处理器
|
||
|
||
# 请求计时（先注册→内层，统计业务耗时）
|
||
@app.middleware("http")
|
||
async def add_process_time_header(request, call_next):
|
||
start_time = time.time()
|
||
response = await call_next(request)
|
||
response.headers["X-Process-Time"] = str(round(time.time() - start_time, 5))
|
||
return response
|
||
|
||
app.middleware("http")(upload_size_limit_middleware) # 上传体积预拦截
|
||
app.middleware("http")(rate_limit_middleware) # 滑动窗口限流
|
||
register_cors(app) # CORS（最后注册→最外层）
|
||
|
||
return app
|
||
```
|
||
|
||
### 新增中间件示例
|
||
|
||
```python
|
||
# src/middleware/custom_middleware.py
|
||
from starlette.requests import Request
|
||
|
||
|
||
async def custom_middleware(request: Request, call_next):
|
||
"""
|
||
自定义中间件：在请求处理前后插入自定义逻辑
|
||
|
||
功能说明：
|
||
1. 请求前：向 request.state 注入自定义数据，供后续业务逻辑使用
|
||
2. 请求后：向响应头添加自定义 Header，用于客户端识别或追踪
|
||
|
||
Args:
|
||
request (Request): Starlette 请求对象，包含请求的所有信息
|
||
call_next: 下一个中间件或路由处理函数
|
||
|
||
Returns:
|
||
Response: 经过处理的响应对象
|
||
"""
|
||
# 【请求前逻辑】在业务处理之前执行，可用于数据预处理、权限预检等
|
||
# 将自定义数据挂载到 request.state，后续路由或中间件可通过 request.state.custom_data 获取
|
||
request.state.custom_data = "xxx"
|
||
|
||
# 调用下一个中间件或最终的路由处理函数，获取响应对象
|
||
response = await call_next(request)
|
||
|
||
# 【响应后逻辑】在响应返回客户端之前执行，可用于添加响应头、日志记录等
|
||
# 添加自定义响应头，客户端可通过 X-Custom-Header 获取该值
|
||
response.headers["X-Custom-Header"] = "value"
|
||
|
||
# 返回最终响应给客户端
|
||
return response
|
||
```
|
||
|
||
```python
|
||
# src/core/app.py
|
||
from middleware.custom_middleware import custom_middleware
|
||
|
||
def create_app() -> FastAPI:
|
||
# ... 其他中间件 ...
|
||
app.middleware("http")(custom_middleware)
|
||
# ...
|
||
```
|
||
|
||
### 中间件执行顺序
|
||
|
||
```
|
||
请求进入 → CORS → 限流 → 上传限制 → 计时 → Token续签 → 操作日志 → DB会话 → Redis注入 → 路由处理
|
||
│
|
||
响应返回 ← CORS ← 限流 ← 上传限制 ← 计时 ← Token续签 ← 操作日志 ← DB会话 ← Redis注入 ← 路由处理 ←┘
|
||
```
|
||
|
||
:::warning 中间件注册顺序
|
||
Starlette 中间件按注册顺序的**反序**执行。最后注册的中间件最先处理请求（最外层）。因此：
|
||
- `register_cors` 最后注册 → 最外层（最先处理请求）
|
||
- `redis_middleware` 最先注册 → 最内层（最后处理请求）
|
||
:::
|
||
|
||
## 四、装饰器扩展
|
||
|
||
### 自定义装饰器
|
||
|
||
通过 `permission_required` 和 `check_demo` 的模式，可以创建自定义装饰器：
|
||
|
||
```python
|
||
# src/common/decorators.py
|
||
|
||
# 自定义装饰器示例：操作日志记录
|
||
def log_operation(module: str, action: str):
|
||
"""
|
||
操作日志记录装饰器
|
||
自动记录接口调用的模块、操作类型、执行耗时
|
||
|
||
Args:
|
||
module: 所属模块名称，如 "培训管理"
|
||
action: 操作类型，如 "新增"、"编辑"、"删除"
|
||
"""
|
||
def decorator(func):
|
||
@wraps(func)
|
||
async def wrapper(*args, **kwargs):
|
||
# 记录开始时间，用于计算接口执行耗时
|
||
start = time.time()
|
||
# 执行原业务函数
|
||
result = await func(*args, **kwargs)
|
||
# 计算耗时（秒）
|
||
duration = time.time() - start
|
||
# 异步记录操作日志到数据库（包含模块、操作、耗时等信息）
|
||
save_log(module, action, duration)
|
||
return result
|
||
return wrapper
|
||
return decorator
|
||
|
||
|
||
# src/modules/system/training/router.py
|
||
|
||
# 使用示例：多个装饰器叠加使用
|
||
# 执行顺序：从下往上执行（先 log_operation，再 check_demo，再 permission_required）
|
||
@router.post('/add')
|
||
@permission_required("sys:training:add") # 权限校验：需要培训新增权限
|
||
@check_demo # 演示环境拦截：检查是否允许写操作
|
||
@log_operation("培训管理", "新增") # 操作日志记录：记录本次操作
|
||
def add(request: Request, data: TrainingForm):
|
||
"""新增培训记录接口"""
|
||
return training_service.add(request, data)
|
||
```
|
||
|
||
## 五、配置扩展
|
||
|
||
### 环境变量驱动
|
||
|
||
所有可配置项通过 `.env` 文件管理，新增配置项只需：
|
||
|
||
1. 在 `.env` 文件中添加变量
|
||
2. 在 `src/core/config/` 对应模块中读取
|
||
3. 在业务代码中引用
|
||
|
||
```bash
|
||
# .env
|
||
CUSTOM_FEATURE_ENABLED=true
|
||
CUSTOM_TIMEOUT=30
|
||
```
|
||
|
||
```python
|
||
# src/core/config/custom.py
|
||
CUSTOM_FEATURE_ENABLED = os.getenv("CUSTOM_FEATURE_ENABLED", "false").lower() == "true"
|
||
CUSTOM_TIMEOUT = int(os.getenv("CUSTOM_TIMEOUT", "30"))
|
||
```
|
||
|
||
## 六、前端扩展
|
||
|
||
### 新增页面
|
||
|
||
1. 在 `ui/src/api/` 创建 API 文件
|
||
2. 在 `ui/src/views/` 创建页面组件
|
||
3. 在数据库 `fastapi_menu` 表中配置菜单路由
|
||
|
||
### 新增组件
|
||
|
||
公共组件放在 `ui/src/components/` 目录，遵循 Vue3 组件规范。
|
||
|
||
## 扩展性矩阵
|
||
|
||
| 扩展维度 | 扩展方式 | 修改范围 | 复杂度 |
|
||
|:---|:---|:---|:---|
|
||
| 新增 CRUD 模块 | 创建 4 文件 + 注册路由 | 新增文件，不改现有代码 | 低 |
|
||
| 新增复杂模块 | 继承基类 + 自定义方法 | 新增文件，不改现有代码 | 中 |
|
||
| 切换数据库驱动 | 修改 .env 配置 | 仅改配置 | 低 |
|
||
| 新增中间件 | 创建中间件 + 注册 | 新增文件 + 改 app.py | 低 |
|
||
| 新增装饰器 | 创建装饰器 + 应用 | 新增文件 | 低 |
|
||
| 新增配置项 | .env + config 模块 | 新增配置 | 低 |
|
||
| 新增前端页面 | API + Views + Menu | 新增文件 + 改数据库 | 中 |
|
||
|
||
## 总结
|
||
|
||
通过分层架构、基类模板方法、装饰器、中间件、环境变量配置等机制，实现了高度的可扩展性。新增 CRUD 模块只需 4 个标准文件 + 1 行路由注册，不修改任何现有代码。数据库驱动通过环境变量切换，中间件通过注册机制插拔。这种设计使得系统在功能持续增长的同时，保持代码结构的清晰和稳定。
|