后端三层架构

Linux

后端三层架构

架构

依赖倒置原则：高层模块不应该依赖低层模块，二者都应依赖抽象接口；抽象接口不应该依赖实现，而实现应依赖抽象接口。

架构到底是什么我现在也没太弄清楚，不过按照我目前的认知，架构大体上可以看作是依赖倒置原则在系统层面的体现，从抽象的角度去规范各个部分之间的连接关系和协作边界。比如对于一个后端服务，从接收前端请求到执行业务逻辑再到数据存储，可以把整个流程大致拆分成三层：表现层、业务层、数据层。

层级 抽象角色 实现角色 解耦作用
表现层 控制器接口 Flask 路由函数 解耦 UI 与服务调用
业务层 服务接口（如 IUserService） 实际服务实现 解耦流程逻辑与实现细节
数据层 数据访问接口（DAO） SQLAlchemy / ORM 解耦业务逻辑与存储实现

每一层都可能包含多个模块以实现不同的功能，不过在确定了采用三层架构之后，实现代码的时候就能明确地知道每个功能属于哪一层以及应该怎么被调用，开发一个更大的功能时候也可以合理地进行任务拆分。一旦这套抽象的结构清晰了，哪怕系统变得越来越大，也不容易变得混乱。比如采用三层架构之后，在写 Flask 路由函数的时候就能明确它属于表现层，只负责接收请求和参数校验，不能写业务逻辑、也不能直接操作数据库。它只能调用业务层暴露出来的接口，后者再去处理复杂的逻辑和数据。这样整个项目就很干净、耦合度比较低。

后端三层架构

1
2
3
🔁 请求 -> 表现层（路由函数） 
              ↳ 业务层（服务逻辑处理）
                  ↳ 数据层（持久化）

1. 表现层（Presentation Layer）

表现层用于接收 HTTP 请求（URL、Body、Header）、参数校验（通常使用 Pydantic）、调用业务逻辑（Service 层）、返回标准响应（JSON）。这里不应该写业务逻辑（比如“用户名是否重复”），也不应直接操作数据库。职责就是请求转发器 + 校验网关。

1
2
3
4
5
6
@app.route("/users", methods=["POST"])
def create_user():
    data = request.json
    user_data = UserCreate(**data)  # Pydantic 校验
    user = UserService.create_user(user_data)
    return jsonify(user.model_dump()), 201

2. 业务层（Business Layer / Service）

业务层负责编排流程、控制事务边界，校验权限、状态（如“是否已存在”），调用 DAO 或外部服务（邮件、缓存、日志等），这一层实际执行业务流程。

1
2
3
4
5
6
7
class UserService:
    @staticmethod
    def create_user(user_data: UserCreate) -> UserOut:
        if DBUserDAO.get_by_username(user_data.username):
            raise ValueError("用户名已存在")
        user = DBUserDAO.create(user_data.model_dump())
        return UserOut.model_validate(user)

3. 数据层（Data Layer / DAO）

数据层只负责与数据库、文件、缓存打交道，封装 SQLAlchemy 的查询、插入、删除等操作。这一层不能加任何业务判断，比如“用户名是否重复”这种逻辑必须上移。

1
2
3
4
class UserDAO:
    @staticmethod
    def get_user_by_username(session, username):
        return session.query(User).filter_by(username=username).first()

常见工具与实践

Pydantic

Pydantic 是一个通过 Python 的类型注解来实现特定数据结构自动校验和转换的工具。比如以下是一个 Pydantic 的使用示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
from pydantic import BaseModel, Field
from typing import Optional

# 👇 模型定义，使用 Field 设置元信息和校验规则
class User(BaseModel):
    id: int = Field(..., gt=0, description="用户ID，必须大于0")
    username: str = Field(..., min_length=3, max_length=20, description="用户名，长度3~20")
    email: Optional[str] = Field(default=None, description="可选邮箱地址")
    age: int = Field(default=18, ge=0, le=150, description="年龄，默认18，范围 0~150")

    # ✅ Pydantic v2 默认已启用 from_attributes，用于支持 ORM 映射
    model_config = {
        "from_attributes": True
    }


# ✅ 使用 model_validate 构造模型（推荐方式）
external_data = {
    "id": "100",  # 字符串将自动转换为 int
    "username": "alice",
    "email": "alice@example.com"
}
user = User.model_validate(external_data)

print("✅ 构建后的模型对象：")
print(user)

# ✅ 使用 copy(update={}) 更新字段
updated_user = user.copy(update={"age": 25})

print("\n✅ 使用 .copy() 更新后的对象：")
print(updated_user)

# ✅ 导出为字典（dict），可以排除某些字段
user_dict = updated_user.model_dump(exclude_none=True)

print("\n✅ 导出为 dict（排除 None）：")
print(user_dict)

# ✅ 导出为 JSON 字符串
user_json = updated_user.model_dump_json(indent=2)

print("\n✅ 导出为 JSON：")
print(user_json)

Pydantic 通过使用 BaseModel 继承方式创建模型，通过类型注解确定各个参数的类型。其中，Field 部分并不是必须的，它是用于额外设置字段校验规则（如 gt=0）和元信息（如 description）。通过 model_validate 方法，可以从字典构建生成 Pydantic 对象。在对象生成的过程中，Pydantic 会自动地进行参数的检查和转换。通过 copy(update={}) 的语法，可以修改 Pydantic 对象的值来生成一个新的对象。可以通过 model_dump 和 model_dump_json 方法来将 Pydantic 对象转换为字典和 JSON，在 Web 应用中非常方便。在三层架构中，表现层可以使用 Pydantic 校验请求参数，业务层可以使用 Pydantic 定义输入输出结构。

以一个用户创建的过程为例，我们首先定义 Pydantic 对象模型，该对象由前端发送过来的 JSON 创建，并作为业务逻辑层的参数。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# schemas/user.py
from pydantic import BaseModel, EmailStr, Field

class UserCreate(BaseModel):
    username: str = Field(..., min_length=3)
    email: EmailStr
    password: str = Field(..., min_length=6)

class UserOut(BaseModel):
    id: int
    username: str
    email: EmailStr

    model_config = {
        "from_attributes": True  # 兼容 ORM 对象
    }

定义一个创建用户的路由函数，通过 Pydantic 对象连接表现层和业务逻辑层：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from flask import request, jsonify
from schemas.user import UserCreate
from services.user_service import UserService

@app.route("/users", methods=["POST"])
def create_user():
    # ✅ 使用 Pydantic v2 的 model_validate
    user_data = UserCreate.model_validate(request.json)

    result = UserService.create_user(user_data)

    # ✅ 使用 model_dump 替代 .dict()
    return jsonify(result.model_dump()), 201

在业务逻辑层中，可以基于 Pydantic 对象的信息，执行调用 DAO 进行数据库写入等操作：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# services/user_service.py
from schemas.user import UserCreate, UserOut
from dao.user_dao import UserDAO

class UserService:
    @staticmethod
    def create_user(user_data: UserCreate) -> UserOut:
        user = UserDAO.create(
            username=user_data.username,
            email=user_data.email,
            password=user_data.password
        )
        # ✅ 使用 from_attributes 替代 from_orm（v2 推荐方式）
        return UserOut.model_validate(user)

DAO：数据访问对象层

DAO（Data Access Object）负责把所有与存储逻辑相关的操作集中封装起来。业务层不直接操作数据库，而是通过 DAO 来完成所有存储操作。这样可以让代码职责清晰、易于测试、易于扩展。数据存储的形式可能是文件系统、数据库、或者缓存，当需要使用不同的存储形式时，可以定义一个 DAO 抽象基类，各个存储形式具体实现该抽象基类的接口。这样可以使得业务层只依赖抽象，不关心底层用的是文件、数据库还是其他。如果调用方需要知道你是谁，那这个抽象就失败了。

我们可以定义一个类似于下方的 DAO，来封装与用户有关的数据库底层操作。这样像上方的 UserService 业务逻辑类，就只需要和 DAO 打交道，不需要实际处理数据库的连接写入等操作了。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
# dao/db_user_dao.py
from sqlalchemy.orm import Session
from models.user import User
from utils.db import with_session
from functools import wraps
from database import SessionLocal

def with_session(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        session = SessionLocal()
        try:
            result = func(*args, session=session, **kwargs)
            session.commit()
            return result
        except Exception:
            session.rollback()
            raise
        finally:
            session.close()
    return wrapper

class DBUserDAO:
    @staticmethod
    @with_session
    def get_by_id(user_id: int, session: Session) -> User | None:
        return session.query(User).filter_by(id=user_id).first()

    @staticmethod
    @with_session
    def get_by_username(username: str, session: Session) -> User | None:
        return session.query(User).filter_by(username=username).first()

    @staticmethod
    @with_session
    def create(user_data: dict, session: Session) -> User:
        user = User(**user_data)
        session.add(user)
        session.flush()
        return user

    @staticmethod
    @with_session
    def delete(user_id: int, session: Session) -> bool:
        session.query(User).filter_by(id=user_id).delete()
        return True

在 database.py 进行一些 SQLAlchemy 的连接配置：

1
2
3
4
5
6
7
8
9
10
11
12
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

DATABASE_URL = "sqlite:///./app.db"

engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False} if "sqlite" in DATABASE_URL else {}
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

在 models/user.py 中定义 ORM 模型：

1
2
3
4
5
6
7
8
9
10
11
# models/user.py
from sqlalchemy import Column, Integer, String
from database import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(100), unique=True, nullable=False)
    password = Column(String(100), nullable=False)

小结

架构类似于一种职责划分机制，它让系统中每一块代码都有清晰的定位，知道自己该做什么、不该做什么。在这篇文章中，我们把“架构”理解为一种更高层次的依赖倒置原则：上层不依赖下层的实现细节，而是依赖于稳定的抽象。架构的核心目标是解耦与稳定，只有当各部分之间建立在抽象接口上的协作关系时，系统才能在不断迭代中保持可维护性。三层架构通过将系统分为表现层、业务层、数据层，每一层各司其职，避免了逻辑混杂和功能重叠，使系统在结构上变得更加清晰、可控、可演进。三层架构在绝大多数中小型 Web 应用中都是非常适用的，比如它特别适合Web 服务型系统（如 Flask、Django、Spring Boot 等）或者API 驱动的接口后端（配合前端或移动端使用）。对于某些高性能系统、事件驱动架构、或者需要大量异步通信的微服务系统，三层架构可能会显得不够灵活，但掌握它是构建一个清晰、稳定、可维护系统的第一步。