分层架构设计

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

架构概览

架构分层原则

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

┌─────────────────────────────────────────────────────────────┐
│                    HTTP 层 (Handler)                         │
│            src/modules/{module}/handlers.py                  │
│  职责：路由定义 · 权限装饰器 · 操作日志 · 响应返回            │
│  原则：不含业务逻辑，仅做请求转发                              │
└──────────────────────────┬──────────────────────────────────┘
                           │ 调用
                           ▼
┌─────────────────────────────────────────────────────────────┐
│                   业务层 (Service)                           │
│            src/modules/{module}/service.py                   │
│  职责：业务逻辑 · 校验规则 · 字段组装 · 序列化加工             │
│  原则：不直接操作数据库，通过 Repository 访问数据               │
└──────────────────────────┬──────────────────────────────────┘
                           │ 调用
                           ▼
┌─────────────────────────────────────────────────────────────┐
│                  数据层 (Repository)                         │
│            src/modules/{module}/repository.py                │
│  职责：数据库 CRUD · 查询条件组装 · 软删除过滤                 │
│  原则：只做数据访问原语，不做业务校验和响应封装                  │
└──────────────────────────┬──────────────────────────────────┘
                           │
                           ▼
┌─────────────────────────────────────────────────────────────┐
│                  Model 层 (ORM 模型)                         │
│            src/modules/{module}/models.py                    │
│  职责：表结构定义 · 字段映射 · 序列化方法                      │
└──────────────────────────┬──────────────────────────────────┘
                           │
                           ▼
┌─────────────────────────────────────────────────────────────┐
│              数据库 (MySQL / PostgreSQL / etc.)               │
└─────────────────────────────────────────────────────────────┘

各层详细职责

第一层：HTTP 层（Handler）

目录位置：src/modules/{module}/handlers.py

核心职责：

权限控制：通过 @permission_required 装饰器声明所需权限
演示模式保护：通过 @check_demo 装饰器阻止演示环境的写操作
操作日志：通过 @operation_log 装饰器记录写操作日志
响应返回：直接返回 Service 层的 R 对象

设计原则：

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

代码示例（岗位模块）：

python

# src/modules/position/handlers.py
# +======================================================================
# | 模块: 岗位请求处理器层
# | 说明: 承接岗位相关的 HTTP 请求，校验权限/演示模式并记录操作日志后，
# |       转交岗位服务层处理
# +======================================================================
"""岗位请求处理器。"""

from common.middleware.access_decorators import check_demo, permission_required
from common.middleware.operation_log import operation_log
from core import response as R
from core.base_handler import BaseHandler
from modules.position.service import position_service


# ======================================================================
# 查询岗位分页数据
# ======================================================================
class PositionPageHandler(BaseHandler):
    """查询岗位分页数据。"""

    @permission_required("sys:position:list")
    async def get(self):
        """GET 请求：返回岗位分页列表。"""
        return await position_service.get_page(self)


# ======================================================================
# 查询岗位详情
# ======================================================================
class PositionDetailHandler(BaseHandler):
    """查询岗位详情。"""

    @permission_required("sys:position:detail")
    async def get(self, id):
        """GET 请求：按 ID 返回岗位详情。

        :param id: 岗位 ID（路径参数）
        """
        return R.ok(self, data=position_service.get_detail(id))


# ======================================================================
# 添加岗位
# ======================================================================
class PositionAddHandler(BaseHandler):
    """添加岗位。"""

    @permission_required("sys:position:add")
    @check_demo
    @operation_log("岗位管理", "添加")
    async def post(self):
        """POST 请求：新增岗位。"""
        return await position_service.add(self)


# ======================================================================
# 更新岗位
# ======================================================================
class PositionUpdateHandler(BaseHandler):
    """更新岗位。"""

    @permission_required("sys:position:update")
    @check_demo
    @operation_log("岗位管理", "更新")
    async def put(self):
        """PUT 请求：更新岗位信息。"""
        return await position_service.update(self)


# ======================================================================
# 删除岗位
# ======================================================================
class PositionDeleteHandler(BaseHandler):
    """删除岗位。"""

    @permission_required("sys:position:delete")
    @check_demo
    @operation_log("岗位管理", "删除")
    async def delete(self, id):
        """DELETE 请求：按 ID 删除岗位（软删除）。

        :param id: 岗位 ID（路径参数）
        """
        return await position_service.delete(self, id)


# ======================================================================
# 设置岗位状态
# ======================================================================
class PositionStatusHandler(BaseHandler):
    """设置岗位状态。"""

    @permission_required("sys:position:status")
    @check_demo
    @operation_log("岗位管理", "设置状态")
    async def put(self):
        """PUT 请求：修改岗位状态（如启用 / 禁用）。"""
        return await position_service.update_status(self)


# ======================================================================
# 获取岗位下拉列表
# ======================================================================
class PositionGetListHandler(BaseHandler):
    """获取岗位下拉列表。"""

    async def get(self):
        """GET 请求：返回岗位下拉列表（无需权限校验）。"""
        return await position_service.get_options(self)


# ======================================================================
# 批量删除岗位
# ======================================================================
class PositionBatchDeleteHandler(BaseHandler):
    """批量删除岗位。"""

    @permission_required("sys:position:delete")
    @check_demo
    @operation_log("岗位管理", "批量删除")
    async def post(self):
        """POST 请求：批量删除岗位（软删除）。"""
        return await position_service.batch_delete(self)

装饰器顺序

装饰器按从外到内执行：先 @permission_required（鉴权），再 @check_demo（演示模式），最后 @operation_log（操作日志）。所有 Handler 方法都是 async def，因为 Tornado Handler 方法默认异步。

第二层：业务层（Service）

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

核心职责：

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

设计原则：

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

代码示例（岗位模块）：

python

# src/modules/position/service.py
# +======================================================================
# | 模块: 岗位业务逻辑层
# | 说明: 定义岗位服务，声明差异点以复用通用 CRUD 流程，并导出服务单例
# +======================================================================
"""岗位业务逻辑。"""

from core.base_service import BaseService
from modules.position.models import Position
from modules.position.repository import PositionRepository
from modules.position.schemas import PositionSchema, PositionStatusSchema


# ======================================================================
# 岗位服务
# ======================================================================
class PositionService(BaseService):
    """岗位服务。"""

    # ============================================================
    # 基础声明
    # ============================================================
    model = Position
    repo_cls = PositionRepository

    # ============================================================
    # 差异点声明
    # ============================================================
    # 校验模型
    create_schema = PositionSchema
    update_schema = PositionSchema
    status_schema = PositionStatusSchema
    # 分页查询字段
    page_like_fields = ("name",)
    page_eq_fields = ("status",)
    # 分页排序与下拉过滤
    page_order_by = (("sort", "asc"),)
    options_filter = {"status": 1}


# ======================================================================
# 服务单例
# ======================================================================
position_service = PositionService()

Service 层的两种模式

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

第三层：数据层（Repository）

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

核心职责：

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

设计原则：

只做数据访问原语，不做 R 封装
等值方法自动软删过滤，filter() 为原始出口
Service 层通过 repo_cls 关联 Repository 类，基类自动实例化

代码示例（岗位模块）：

python

# src/modules/position/repository.py
# +======================================================================
# | 模块: 岗位数据访问层
# | 说明: 定义岗位仓储，继承 BaseRepository 获得通用 CRUD 与软删除能力
# +======================================================================
"""岗位数据访问。"""

from core.base_repository import BaseRepository
from modules.position.models import Position


# ======================================================================
# 岗位仓储
# ======================================================================
class PositionRepository(BaseRepository):
    """岗位仓储。"""

    # ============================================================
    # 绑定的模型类
    # ============================================================
    model = Position

Repository 基类提供的方法

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

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

第四层：Model 层（ORM 模型）

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

核心职责：

表结构定义：通过 SQLAlchemy Column 定义字段
通用字段继承：BaseModel 提供 id/create_user/create_time/update_user/update_time/is_delete
序列化方法：to_dict() 自动排除敏感字段（password/salt/is_delete），驼峰命名

代码示例（岗位模块）：

python

# src/modules/position/models.py
from sqlalchemy import Column, Integer, String, text

from config.database import DB_PREFIX
from core.base_model import BaseModel


class Position(BaseModel):
    """岗位模型。"""
    __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="岗位状态")
    # 岗位排序
    sort = Column(Integer, default=0, server_default=text('0'), comment="岗位排序")

Schema 层（请求校验）

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

核心职责：

请求参数校验：使用 Pydantic v2 的 BaseModel + Field 声明约束
创建/编辑共用：继承 BaseSchema（自带可选 id 字段）
状态更新独立：单独建类，只含 id + status

代码示例（岗位模块）：

python

# src/modules/position/schemas.py
from pydantic import BaseModel, Field

from core.base_schema import BaseSchema


class PositionSchema(BaseSchema):
    """岗位创建/编辑表单。"""
    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 PositionStatusSchema(BaseModel):
    """岗位状态更新表单。"""
    id: int = Field(..., gt=0, description="岗位ID")
    status: int = Field(..., ge=1, le=2, description="岗位状态：1-在用 2-停用")

路由注册

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

每个模块在 src/api/v1/ 下定义路由文件，导出 routes 列表，由 router.py 统一聚合。

代码示例（岗位模块）：

python

# src/api/v1/position.py
"""岗位相关路由。"""

from modules.position.handlers import (
    PositionAddHandler, PositionBatchDeleteHandler, PositionDeleteHandler,
    PositionDetailHandler, PositionGetListHandler, PositionPageHandler,
    PositionStatusHandler, PositionUpdateHandler,
)

routes = [
    (r"/position/page", PositionPageHandler),
    (r"/position/detail/(\d+)", PositionDetailHandler),
    (r"/position/getPositionList", PositionGetListHandler),
    (r"/position/add", PositionAddHandler),
    (r"/position/update", PositionUpdateHandler),
    (r"/position/delete/(\d+)", PositionDeleteHandler),
    (r"/position/status", PositionStatusHandler),
    (r"/position/batchDelete", PositionBatchDeleteHandler),
]

router.py 中统一注册：

python

# src/api/v1/router.py
from api.v1 import position  # 导入模块

_RESOURCE_MODULES = [
    # ... 其他模块 ...
    position,  # 注册
]

def build_routes() -> list:
    routes = []
    for module in _RESOURCE_MODULES:
        routes.extend([(V1_PREFIX + path, handler) for path, handler in module.routes])
    routes.append((V1_PREFIX + "/health", HealthHandler))
    return routes

依赖方向

HTTP 层 (Handler) ──依赖──→ Service 层 ──依赖──→ Repository 层 ──依赖──→ Model
       │                        │                      │
       │                        │                      │
       ▼                        ▼                      ▼
  R (响应封装)              R (响应封装)          BaseRepository (基类)
  装饰器(BaseHandler)     BaseRepository         SQLAlchemy ORM
  access_decorators       Pydantic Schema

依赖规则

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

完整调用链路示例

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

POST /api/v1/position/add
│
├─ 1. Handler (handlers.py)
│    @permission_required("sys:position:add")
│    @check_demo
│    @operation_log("岗位管理", "添加")
│    async def post(self):
│        return await position_service.add(self)
│
├─ 2. Service (service.py → base_service.py)
│    async def add(self, handler):
│        form = self._validate(handler, self.create_schema)  # Pydantic 校验
│        err = self._check_unique(form)                      # 唯一性校验
│        self._before_add(handler, data)                     # 前置钩子
│        fields = self._build_create_fields(data)            # 字段组装
│        self.repo.create(self.model(**fields))              # 调用 Repository
│        return R.ok(handler, msg="添加成功")
│
├─ 3. Repository (base_repository.py)
│    def create(self, obj):
│        obj.create_user = str(self._uid())   # 自动写入创建人
│        self.db.add(obj)
│        self.db.commit()
│        return obj
│
├─ 4. Model (base_model.py)
│    class Position(BaseModel):
│        __tablename__ = "tornado_position"
│        name = Column(String(255), ...)
│
└─ 5. Database
     INSERT INTO tornado_position (name, status, sort, create_user, create_time)
     VALUES ('高级工程师', 1, 10, '1', '2026-09-16 10:00:00')

各层文件职责一览

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

总结

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

🔐 认证与安全

📊 数据与配置

📝 日志与监控

🛡️ 防护与治理

📁 文件处理

🔧 核心工具

🧩 辅助工具

分层架构设计

架构概览

各层详细职责

第一层：HTTP 层（Handler）

第二层：业务层（Service）

第三层：数据层（Repository）

第四层：Model 层（ORM 模型）

Schema 层（请求校验）

路由注册

依赖方向

完整调用链路示例

各层文件职责一览

总结

Become a sponsor

企业级中台解决方案

企业级中台解决方案

分层架构设计 ​

架构概览 ​

各层详细职责 ​

第一层：HTTP 层（Handler） ​

第二层：业务层（Service） ​

第三层：数据层（Repository） ​

第四层：Model 层（ORM 模型） ​

Schema 层（请求校验） ​

路由注册 ​

依赖方向 ​

完整调用链路示例 ​

各层文件职责一览 ​

总结 ​

分层架构设计

架构概览

各层详细职责

第一层：HTTP 层（Handler）

第二层：业务层（Service）

第三层：数据层（Repository）

第四层：Model 层（ORM 模型）

Schema 层（请求校验）

路由注册

依赖方向

完整调用链路示例

各层文件职责一览

总结