扩展性设计

本章详细描述系统的扩展性设计，包括模块可插拔、数据库驱动可切换、中间件可插拔三大扩展维度。

扩展性设计原则

遵循开闭原则（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

驱动切换注意事项

切换驱动后需安装对应的 Python 包（如 psycopg[binary] for PostgreSQL）
数据迁移使用 scripts/migrate_db.py 工具
SQLite 不支持并发写入，仅适用于开发/测试环境
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() 中按顺序注册。新增中间件只需：

在 src/middleware/ 目录创建中间件文件
在 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注入 ← 路由处理 ←┘

中间件注册顺序

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 文件管理，新增配置项只需：

在 .env 文件中添加变量
在 src/core/config/ 对应模块中读取
在业务代码中引用

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"))

六、前端扩展

新增页面

在 ui/src/api/ 创建 API 文件
在 ui/src/views/ 创建页面组件
在数据库 fastapi_menu 表中配置菜单路由

新增组件

公共组件放在 ui/src/components/ 目录，遵循 Vue3 组件规范。

扩展性矩阵

扩展维度	扩展方式	修改范围	复杂度
新增 CRUD 模块	创建 4 文件 + 注册路由	新增文件，不改现有代码	低
新增复杂模块	继承基类 + 自定义方法	新增文件，不改现有代码	中
切换数据库驱动	修改 .env 配置	仅改配置	低
新增中间件	创建中间件 + 注册	新增文件 + 改 app.py	低
新增装饰器	创建装饰器 + 应用	新增文件	低
新增配置项	.env + config 模块	新增配置	低
新增前端页面	API + Views + Menu	新增文件 + 改数据库	中

总结

通过分层架构、基类模板方法、装饰器、中间件、环境变量配置等机制，实现了高度的可扩展性。新增 CRUD 模块只需 4 个标准文件 + 1 行路由注册，不修改任何现有代码。数据库驱动通过环境变量切换，中间件通过注册机制插拔。这种设计使得系统在功能持续增长的同时，保持代码结构的清晰和稳定。

🔐 认证与安全

📊 数据与配置

📝 日志与监控

🛡️ 防护与治理

📁 文件处理

🔧 核心工具

🧩 辅助工具

扩展性设计

一、模块可插拔设计

模块标准结构

新增模块步骤

模块解耦

复杂模块扩展

二、数据库驱动可切换

驱动注册机制

切换方式

跨库数据迁移

三、中间件可插拔

中间件注册机制

新增中间件示例

中间件执行顺序

四、装饰器扩展

自定义装饰器

五、配置扩展

环境变量驱动

六、前端扩展

新增页面

新增组件

扩展性矩阵

总结

Become a sponsor

企业级中台解决方案

企业级中台解决方案

扩展性设计 ​

一、模块可插拔设计 ​

模块标准结构 ​

新增模块步骤 ​

模块解耦 ​

复杂模块扩展 ​

二、数据库驱动可切换 ​

驱动注册机制 ​

切换方式 ​

跨库数据迁移 ​

三、中间件可插拔 ​

中间件注册机制 ​

新增中间件示例 ​

中间件执行顺序 ​

四、装饰器扩展 ​

自定义装饰器 ​

五、配置扩展 ​

环境变量驱动 ​

六、前端扩展 ​

新增页面 ​

新增组件 ​

扩展性矩阵 ​

总结 ​

扩展性设计

一、模块可插拔设计

模块标准结构

新增模块步骤

模块解耦

复杂模块扩展

二、数据库驱动可切换

驱动注册机制

切换方式

跨库数据迁移

三、中间件可插拔

中间件注册机制

新增中间件示例

中间件执行顺序

四、装饰器扩展

自定义装饰器

五、配置扩展

环境变量驱动

六、前端扩展

新增页面

新增组件

扩展性矩阵

总结