分层架构设计

本章详细描述的三层架构设计，包括各层的职责边界、依赖方向、通信方式，并以岗位（Position）模块为实例贯穿说明。

三层架构概览

架构分层原则

采用经典的三层架构，核心原则是：依赖单向、职责单一、接口隔离。上层只依赖下层接口，下层不感知上层存在。

┌─────────────────────────────────────────────────────────────┐
│ HTTP 层 (Endpoint) │
│ api/v1/endpoints/position.py │
│ 职责：路由定义 · 权限装饰器 · 参数接收 · 响应返回 │
│ 原则：不含业务逻辑，仅做请求转发 │
└──────────────────────────┬──────────────────────────────────┘
 │ 调用
 ▼
┌─────────────────────────────────────────────────────────────┐
│ 业务层 (Service) │
│ modules/system/position/service.py │
│ 职责：业务逻辑 · 校验规则 · 字段组装 · 序列化加工 │
│ 原则：不直接操作数据库，通过 Repository 访问数据 │
└──────────────────────────┬──────────────────────────────────┘
 │ 调用
 ▼
┌─────────────────────────────────────────────────────────────┐
│ 数据层 (Repository) │
│ modules/system/position/repository.py │
│ 职责：数据库 CRUD · 查询条件组装 · 软删除过滤 │
│ 原则：只做数据访问原语，不做业务校验和响应封装 │
└──────────────────────────┬──────────────────────────────────┘
 │
 ▼
┌─────────────────────────────────────────────────────────────┐
│ 数据库 (MySQL / PostgreSQL / etc.) │
└─────────────────────────────────────────────────────────────┘

各层详细职责

第一层：HTTP 层（Endpoint）

目录位置：src/api/v1/endpoints/{module}.py

核心职责：

路由定义：声明 HTTP 方法、URL 路径、请求参数类型
权限控制：通过 @permission_required 装饰器声明所需权限
演示模式保护：通过 @check_demo 装饰器阻止演示环境的写操作
参数接收：通过 Pydantic 模型自动校验请求体
响应返回：直接返回 Service 层的 R 对象

设计原则：

不包含任何业务逻辑
不直接访问数据库
仅做请求转发和响应返回
权限字符串遵循 sys:{module}:{action} 格式

代码示例（岗位模块）：

python

# src/api/v1/endpoints/position.py
# +======================================================================
# | 模块: 岗位路由层
# | 说明: 提供岗位信息的增删改查、状态管理、列表获取等 API 接口
# +======================================================================

from fastapi import APIRouter, Request

from modules.system.position.schemas import PositionForm, PositionStatusForm
from core.access_decorators import check_demo
from core.access_decorators import permission_required
from modules.system.position.service import position_service

from core import R

# ============================================================
# 创建路由实例
# ============================================================
router = APIRouter()


# ============================================================
# 查询分页数据
# ============================================================
@router.get('/page', summary='查询分页数据')
@permission_required("sys:position:page")
def page(request: Request):
    """
    获取岗位分页列表
    :param request: FastAPI请求对象
    :return: 分页数据响应
    """
    return position_service.get_page(request)


# ============================================================
# 查询岗位详情
# ============================================================
@router.get('/detail/{id}', summary='查询岗位详情')
@permission_required("sys:position:detail")
def detail(request: Request, id):
    """
    根据岗位ID查询详细信息
    :param request: FastAPI请求对象
    :param id: 岗位ID（路径参数）
    :return: 岗位详情数据
    """
    return R.ok(data=position_service.get_detail(id))


# ============================================================
# 获取岗位列表
# ============================================================
@router.get('/list', summary='获取岗位列表')
@permission_required("sys:position:list")
def get_position_list(request: Request):
    """
    获取所有岗位列表（无分页，用于下拉选择）
    :param request: FastAPI请求对象
    :return: 岗位列表数据
    """
    return R.ok(data=position_service.get_position_list(request))


# ============================================================
# 添加岗位
# ============================================================
@router.post('/add', summary='添加岗位')
@permission_required("sys:position:add")
@check_demo
def add(request: Request, data: PositionForm):
    """
    新增岗位
    :param request: FastAPI请求对象
    :param data: 岗位表单数据
    :return: 添加结果响应
    """
    return position_service.add(request, data)


# ============================================================
# 更新岗位
# ============================================================
@router.put('/update', summary='更新岗位')
@permission_required("sys:position:update")
@check_demo
def update(request: Request, data: PositionForm):
    """
    更新岗位信息
    :param request: FastAPI请求对象
    :param data: 岗位表单数据（必须包含 id 字段）
    :return: 更新结果响应
    """
    return position_service.update(request, data)


# ============================================================
# 删除岗位
# ============================================================
@router.delete('/delete/{id}', summary='删除岗位')
@permission_required("sys:position:delete")
@check_demo
def delete(request: Request, id):
    """
    根据ID删除单个岗位
    :param request: FastAPI请求对象
    :param id: 岗位ID（路径参数）
    :return: 删除结果响应
    """
    return position_service.delete(id)


# ============================================================
# 设置岗位状态
# ============================================================
@router.put('/status', summary='设置岗位状态')
@permission_required("sys:position:status")
@check_demo
def status(request: Request, data: PositionStatusForm):
    """
    修改岗位状态（启用/禁用）
    :param request: FastAPI请求对象
    :param data: 状态表单数据
    :return: 状态更新结果响应
    """
    return position_service.update_status(request, data)


# ============================================================
# 批量删除岗位
# ============================================================
@router.delete('/batchDelete', summary='批量删除岗位')
@permission_required("sys:position:batchDelete")
@check_demo
async def batch_delete(request: Request):
    """
    批量删除岗位（请求体需传入 ID 数组）

    Request Body 示例:
        [1, 2, 3]  # 岗位ID数组

    :param request: FastAPI请求对象
    :return: 批量删除结果响应
    """
    return await position_service.batch_delete(request)

装饰器顺序

装饰器按从上到下执行：先 @permission_required（鉴权），再 @check_demo（演示模式），最后才是路由函数。批量删除端点为 async def，因为需要 await parse_batch_ids(request) 异步读取请求体。

第二层：业务层（Service）

目录位置：src/modules/{group}/{name}/service.py

核心职责：

业务逻辑编排：组合多个数据操作完成业务流程
唯一性校验：通过 unique_fields 声明唯一约束
字段组装：_build_create_fields / _build_update_fields 组装入库字段
序列化加工：_serialize / _serialize_detail 加工响应数据
文件处理：文件字段迁移、URL 补全
前置/后置钩子：_before_add / _before_update / _before_delete 扩展点

设计原则：

通过 Repository 访问数据库，不直接使用 db.query()
返回 R 对象，不做 JSONResponse 封装
通过声明差异点定制通用 CRUD，减少重复代码

代码示例（岗位模块）：

python

# src/modules/system/position/service.py
# +======================================================================
# | 模块: 岗位业务逻辑层
# | 说明: 岗位的增删改查、唯一性校验、状态管理等业务处理
# +======================================================================

from typing import Optional

from core.base_service import BaseService
from modules.system.position.models import Position
from modules.system.position.repository import position_repo
from modules.system.user.models import User
from modules.system.user.repository import user_repo
from utils.request import parse_id_list


# ============================================================
# 岗位业务服务类
# ============================================================
class PositionService(BaseService[Position]):
    """岗位业务服务类，继承基础服务获得通用 CRUD 能力"""
    # 数据访问与模型
    repo = position_repo
    model = Position
    # 分页差异点
    page_like_fields = ('name',)
    page_eq_fields = ('status',)
    page_order_by = (('sort', 'asc'),)
    # 唯一性校验
    unique_fields = {'name': '岗位名称不能重复'}

    # ============================================================
    # 删除前校验
    # ============================================================
    def _before_delete(self, ids) -> Optional[str]:
        """删除前校验：存在用户引用该岗位时禁止删除，避免用户岗位悬空"""
        id_list = parse_id_list(str(ids))
        if id_list and user_repo.filter(User.position_id.in_(id_list), User.is_delete == 0).first():
            return "存在用户引用该岗位，请先调整用户岗位"
        return None

    # ============================================================
    # 获取岗位数据列表
    # ============================================================
    def get_position_list(self, request):
        """获取岗位数据列表"""
        # 与 /page 一致的筛选条件：name 模糊、status 精确；保留软删过滤与排序
        query = self._apply_page_filters(self.repo.filter_by(), request)
        query = query.order_by(self.model.sort.asc())
        # 返回结果
        return [v.to_dict() for v in query.all()]


# ============================================================
# 模块级单例
# ============================================================
position_service = PositionService()

Service 层的两种模式

简单 CRUD 模块继承 BaseService，只需声明差异点即可获得完整的增删改查能力。复杂模块（如用户、角色、菜单）在继承基础上添加自定义方法，或完全自定义 Service 函数。

第三层：数据层（Repository）

目录位置：src/modules/{group}/{name}/repository.py

核心职责：

CRUD 操作：create / update / batch_delete / get_by_id 等
查询构建：filter / filter_by / get_one / get_all / count
分页查询：paginate 返回 (items, total) 元组
软删除过滤：等值方法自动追加 is_delete=0
数据清洗：_clean_model_data 将 camelCase 转 snake_case

设计原则：

只做数据访问原语，不做 R 封装
等值方法自动软删过滤，filter() 为原始出口
删除的 R 响应由 Service 层 batch_delete_with_r 统一产出

代码示例（岗位模块）：

python

# src/modules/system/position/repository.py
# +======================================================================
# | 模块: 岗位数据访问层
# | 说明: 封装岗位模型的数据访问操作
# +======================================================================

"""modules/system/position 数据访问层。"""
from modules.system.position.models import Position
from core.base_repository import BaseRepository


# ============================================================
# 岗位仓库类
# ============================================================
class PositionRepository(BaseRepository[Position]):
    """岗位数据仓库，继承基础仓库获得通用 CRUD 能力"""
    pass


# ============================================================
# 仓库实例
# ============================================================
position_repo = PositionRepository(Position)

Repository 基类提供的方法

BaseRepository 提供以下通用方法，子类可直接使用：

查询：get_by_id / get_by_ids / get_one / get_all / count / exists / exists_by_field
分页：paginate 返回 (items, total)
写操作：create / update / batch_delete
条件构建：filter（原始出口） / filter_by（等值出口）

依赖方向

HTTP 层 ──依赖──→ Service 层 ──依赖──→ Repository 层 ──依赖──→ Model
 │ │ │
 │ │ │
 ▼ ▼ ▼
R (响应封装) R (响应封装) BaseRepository (基类)
Pydantic BaseRepository get_db_session()
AccessDecorator BaseService SQLAlchemy ORM

依赖规则

HTTP 层只依赖 Service 层，不直接调用 Repository
Service 层通过 Repository 访问数据库，不直接使用 db.query()
Repository 层只依赖 Model 和数据库会话
下层不感知上层存在，不调用上层方法

模型层（Model）

目录位置：src/modules/{group}/{name}/models.py

模型层定义数据库表结构，继承 base_model（通用字段）和 base_db（持久化方法）。

python

# src/modules/system/position/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 Position(base_model, base_db):
    """岗位模型类"""
    # ============================================================
    # 表名配置
    # ============================================================
    __tablename__ = DB_PREFIX + "position"
    __table_comment__ = "岗位表"

    # ============================================================
    # 字段定义
    # ============================================================
    # 岗位名称
    name = Column(String(255), nullable=False, index=True, comment="岗位名称")
    # 岗位状态：1-在用 2-停用
    status = Column(Integer, default=0, server_default=text('0'), index=True, comment="岗位状态：1-在用 2-停用")
    # 岗位排序
    sort = Column(Integer, default=0, server_default=text('0'), comment="岗位排序")

    # ============================================================
    # 内置方法
    # ============================================================
    def __repr__(self):
        """返回岗位名称作为字符串表示"""
        return '岗位：{}'.format(self.name)

Schema 层（校验）

目录位置：src/modules/{group}/{name}/schemas.py

Schema 层定义请求参数的校验规则，使用 Pydantic v2 的 BaseModel + Field。

python

# src/modules/system/position/schemas.py
# +======================================================================
# | 模块: 岗位表单验证
# | 说明: 岗位创建/编辑/状态更新的请求数据校验
# +======================================================================

from typing import Optional

from pydantic import BaseModel, Field
from core.base_schemas import BaseForm


# ============================================================
# 岗位表单类
# ============================================================
class PositionForm(BaseForm):
    """岗位创建/编辑表单"""
    name: str = Field(..., min_length=1, max_length=150, description="岗位名称")
    status: int = Field(..., ge=1, le=2, description="岗位状态：1-在用 2-停用")
    sort: int = Field(..., ge=0, le=99999, description="岗位排序")


# ============================================================
# 设置状态表单类
# ============================================================
class PositionStatusForm(BaseModel):
    """岗位状态更新表单"""
    id: int = Field(..., gt=0, description="岗位ID")
    status: int = Field(..., ge=1, le=2, description="岗位状态：1-在用 2-停用")

完整调用链路示例

以添加岗位为例，展示三层之间的调用关系：

POST /api/v1/position/add
│
├─ 1. Endpoint (position.py)
│ @permission_required("sys:position:add")
│ @check_demo
│ def add(request, data: PositionForm):
│ return position_service.add(request, data)
│
├─ 2. Service (service.py)
│ def add(self, request, data):
│ err = self._check_unique(data) # 唯一性校验
│ self._before_add(request, data) # 前置钩子
│ fields = self._build_create_fields() # 字段组装
│ self.repo.create(**fields) # 调用 Repository
│ return R.ok(msg='添加成功')
│
├─ 3. Repository (base_repository.py)
│ def create(self, **fields):
│ obj = self.model(**self._clean_model_data(fields))
│ obj.save() # add + flush
│ return obj
│
├─ 4. Model (base_db.py)
│ def save(self):
│ db = get_db_session()
│ db.add(self)
│ db.flush() # SQL 发送到数据库
│
└─ 5. Database
 INSERT INTO fastapi_position (name, status, sort, create_user, ...)
 VALUES ('高级工程师', 1, 10, '管理员', ...)

各层文件职责一览

文件	层次	职责	是否包含业务逻辑
`endpoints/position.py`	HTTP	路由、权限、参数接收	否
`position/schemas.py`	HTTP	请求参数校验规则	否
`position/service.py`	业务	业务逻辑、校验、序列化	是
`position/repository.py`	数据	数据库 CRUD、查询构建	否
`position/models.py`	数据	表结构定义、字段映射	否

总结

三层架构通过依赖单向和职责单一实现了良好的关注点分离。HTTP 层只关心请求接收和响应返回，Service 层只关心业务逻辑编排，Repository 层只关心数据访问。这种分层使得每一层都可以独立测试、独立替换，系统在模块数量增长时仍能保持清晰的代码结构。以岗位模块为例，完整实现一个 CRUD 模块只需 5 个文件，其中 Repository 和 Model 几乎零代码，开发者只需关注 Service 层的业务差异点。

🔐 认证与安全

📊 数据与配置

📝 日志与监控

🛡️ 防护与治理

📁 文件处理

🔧 核心工具

🧩 辅助工具

分层架构设计

三层架构概览

各层详细职责

第一层：HTTP 层（Endpoint）

第二层：业务层（Service）

第三层：数据层（Repository）

依赖方向

模型层（Model）

Schema 层（校验）

完整调用链路示例

各层文件职责一览

总结

Become a sponsor

企业级中台解决方案

企业级中台解决方案

分层架构设计 ​

三层架构概览 ​

各层详细职责 ​

第一层：HTTP 层（Endpoint） ​

第二层：业务层（Service） ​

第三层：数据层（Repository） ​

依赖方向 ​

模型层（Model） ​

Schema 层（校验） ​

完整调用链路示例 ​

各层文件职责一览 ​

总结 ​

分层架构设计

三层架构概览

各层详细职责

第一层：HTTP 层（Endpoint）

第二层：业务层（Service）

第三层：数据层（Repository）

依赖方向

模型层（Model）

Schema 层（校验）

完整调用链路示例

各层文件职责一览

总结