后端开发架构设计
需求
产品说明
开发一个具有课程管理模块的后端系统，模块功能包含课程的增删改查。
分层介绍
代码分层设计是一种将软件系统按照不同的功能和责任划分为多个层次的方法，通过将不同层次的代码逻辑和功能分开，实现模块化的开发和维护，提高代码的可读性、可维护性和可扩展性。
分层的价值
- 不分层
- 耦合性增加：各个功能和模块之间容易产生强耦合，代码之间的关系复杂且难以理解。这样一来，当需要修改一个功能时，可能会导致其他功能受到不必要的影响，增加了系统的脆弱性。
- 难以维护：缺乏代码分层可能导致代码结构混乱，难以理解和维护。当需要定位和修复 Bug 时，因为缺乏清晰的组织结构，开发人员需要花费更多的时间和精力来查找和修复问题。
- 测试困难：单元测试和集成测试也变得困难没有明确定义的层级，单元测试和集成测试变得困难。不同功能之间的耦合性增加，可能需要进行大量的冗余代码测试和模拟，导致测试代码的复杂性和难度增加。
- 分层
- 可扩展性：代码分层可以使系统更容易扩展。通过将不同的功能划分到不同的层级中，可以避免模块之间的耦合，从而使系统的各个组件更容易被替换或添加新的功能。
- 可维护性：通过将代码逻辑划分到不同的层级中，可以使系统更具可读性和可维护性。每个层级都有自己的职责，使得代码的组织和理解更加清晰，便于团队协作和维护。
- 可测试性：代码分层可以增强系统的可测试性。每个层级都可以被独立地进行单元测试，从而确保每个组件的功能正确性和稳定性。
如何应用分层
当涉及到后端设计分层时，确实需要根据具体项目的实际情况进行引入。实际的项目可能有不同的需求和复杂度，因此，在设计分层架构时需要灵活考虑，并结合项目的实际情况进行调整和优化。
经典三层架构
经典的三层架构是一种将应用程序划分为表示层、业务逻辑层和数据访问层的架构模式。它的重点是分离不同层的职责和关注点，以实现可重用、可维护和可扩展的应用程序设计。
- web/controller 层：这一层主要负责处理用户请求，进行访问控制、基本参数校验，以及执行不太复杂的业务逻辑。通常使用 Web 框架（如 Flask、Django）处理 HTTP 请求。
- Service 层：相对具体的业务逻辑服务层。
- Manager 层：通用业务处理层，它有如下特征：
- 封装对第三方平台的调用，包括预处理返回结果和转化异常信息，以适配上层接口。
- 下沉通用能力，如缓存方案、中间件通用处理，以提供 Service 层的支持。
- 与 DAO 层交互，对多个 DAO 的组合复用。
- DAO 层：DAO 层是数据访问层，用于与底层数据存储（如 MySQL、Oracle、Hbase 等）进行数据交互。
分层领域模型
基于经典三层架构，可以把后端服务按照以下方式进行分层建模。
- DO（Data Object）：此对象与数据库表结构一一对应，通过 DAO 层向上传输数据源对象。
- DAO（Data Access Object）：数据访问对象，负责对数据库的访问和操作，包括CRUD（创建、读取、更新、删除）操作。
- DTO（Data Transfer Object）：数据传输对象，Service 层向外传输的对象。
- Service：服务层，主要负责实现应用程序的业务逻辑。
- Controller：控制器层，主要负责处理客户端的 HTTP 请求，并将请求转发到相应的 Service 层进行业务逻辑处理。
实战思路
接下来使用以下思路完成整个项目的分层开发。
目录结构
项目目录结构可以参考如下设置。
Hogwarts $ tree
.
├── src
├── server.py
├── do/
├── dao/
├── dto/
├── service/
├── controller/
├── tests/
├── /service
封装 do 模型
此对象与数据库表结构一一对应，然后通过 DAO 层向上传输数据源对象。
请注意，此对象中不应包含任何数据库操作。
# do/course_do.py
from typing import Optional
from sqlalchemy import String, Column, Integer
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from platform_bankend.src.server import Base
# 定义orm对象用于操作数据库
class CourseDo(Base):
__tablename__ = "t_course"
id: Mapped[int] = Column(Integer, primary_key=True)
name: Mapped[str] = mapped_column(String(30))
detail: Mapped[Optional[str]] = mapped_column(String(255), default='')
定义表结构时需要集成基类。接下来就需要定义好基类并导入到本文件中。
在服务入口文件中完成基类的声明与数据库连接操作。
server.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase, Session
class Base(DeclarativeBase):
pass
# 配置连接信息
engine = create_engine("mysql+pymysql://root:hogwarts@localhost/course")
# 创建数据库表
Base.metadata.create_all(engine)
# 创建session对象
DBSession = sessionmaker(bind=engine)
db_session: Session = DBSession()
完成表创建。
执行代码，如果没有报错，则创建表成功。可以进入数据库中查看表创建结果。
do/init_db.py
from platform_bankend.src.do.course_do import CourseDo
from platform_bankend.src.server import Base, engine
if __name__ == '__main__':
# 创建表
Base.metadata.create_all(engine)
分离 dao 层
DAO 层是数据访问层，用于与底层数据存储（如 MySQL、Oracle、Hbase 等）进行数据交互。
如果对存储数据有替换的需求，可以考虑将其抽象封装（参考 service 层）。
下面定义 Course 数据模型具体的数据库操作。
这样，对于课程表的基础操作就定义好了。
# dao/course_dao.py
from platform_bankend.src.do.course_do import CourseDo
from platform_bankend.src.server import db_session
class CourseDao:
# 处理数据库相关操作
def get(self, course_id: int) -> CourseDo:
'''
通过 id 查询单条记录
:param course_id: 课程 id
:return: 课程对象
'''
return db_session.query(CourseDo).filter_by(id=course_id).first()
def list(self) -> [CourseDo]:
'''
查询表中所有记录
:return: 课程对象列表
'''
return db_session.query(CourseDo).all()
def create(self, course):
'''
课程表中添加数据
:param course: 课程对象
:return: 课程 id
'''
# 添加到表中
db_session.add(course)
# 提交操作
db_session.commit()
# 返回课程 id
return course.id
def update(self, course_info: dict):
'''
更新课程表中数据
:param course_info: 字典格式，课程信息
:return: 课程 id
'''
# 获取课程 id
course_id = course_info.get("id")
# 通过课程 id 更新课程数据
db_session.query(CourseDo).filter_by(id=course_id).update(course_info)
# 提交操作
db_session.commit()
return course_id
def delete(self, course_id: int):
'''
删除课程表中数据
:param course_id: 课程 id
:return: 课程 id
'''
# 删除课程
db_session.query(CourseDo).filter_by(id=course_id).delete()
# 提交操作
db_session.commit()
return course_id
封装 dto 模型
DTO（Data Transfer Object）：数据传输对象，Service 或 Manager 向外传输的对象。
在这一层使用 pydantic 用于数据验证与解析。
# dto/course_dto.py
from typing import Optional
from pydantic import BaseModel, ConfigDict, constr
# 定义pydantic对象，用于数据验证与解析
class CourseDto(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
name: constr(max_length=30)
detail: Optional[constr(max_length=255)] = ""
分离 service 层
Service 层：相对具体的业务逻辑服务层。
Service 层是整个系统代码最复杂的一层，所以需要对其抽象处理，使其解耦和具备良好的可拓展性、可读、可测性。
新建抽象类，这样就可以完成对于课程增删改查的具体操作。
# service/course_service.py
from platform_bankend.src.dao.course_dao import CourseDao
from platform_bankend.src.do.course_do import CourseDo
# 获取基础数据库操作的实例
course_dao: CourseDao = CourseDao()
class CourseService:
# 实现业务方法，复制的业务逻辑在此处实现
def get(self, course_id: int):
'''
通过课程 id 获取课程信息
:param course_id: 课程 id
:return: 错误信息/字典格式课程信息
'''
# 通过 id 查询课程
course = course_dao.get(course_id)
# 返回字典格式的课程信息
return course
def create(self, course_info: dict):
'''
新增课程
:param course_info: 字典格式，课程信息
:return:
'''
# 为课程信息添加 id
# course_info.update({"id": 0})
print(f"课程信息为 {course_info}")
# 创建课程对象
course = CourseDo(**course_info)
# 新增课程
course_id = course_dao.create(course)
# 返回课程 id
return course_id
def list(self):
'''
获取表中全部数据列表
:return: 字典格式课程信息的列表
'''
course_list = course_dao.list()
return course_list
def update(self, course_info: dict):
'''
更新课程
:param course_info: 字典格式，课程信息
:return: 提示信息/课程 id
'''
# 更新课程
course_id = course_dao.update(course_info)
return course_id
def delete(self, course_id: int):
'''
删除课程
:param course_id: 要删除的课程 id
:return: 提示信息/课程 id
'''
# 删除课程
c_id = course_dao.delete(course_id)
return c_id
下面对 service 当中实现的方法完成单元测试。
当所有的单元测试用例执行通过时，证明 service 中的方法定义没有问题。
# tests/test_course_service.py
from platform_bankend.src.service.course_service import CourseService
class TestCourseService:
def setup_class(self):
self.cs = CourseService()
def test_create(self):
course_info = {
"id": 1,
"name": "ck21",
"detail": "python进阶班"
}
course_id = self.cs.create(course_info)
assert course_id == 1
def test_get(self):
course = self.cs.get(1)
print(course)
assert course.id == 1
def test_list(self):
course_list = self.cs.list()
print(course_list)
course_id_list = [c.id for c in course_list]
print(course_id_list)
assert 1 in course_id_list
def test_update(self):
course_info = {
"id": 1,
"name": "测试开发班",
"detail": "python进阶班update"
}
course_id = self.cs.update(course_info)
assert course_id == 1
def test_delete(self):
course_id = self.cs.delete(3)
course_list = self.cs.list()
print(course_list)
course_id_list = [c.id for c in course_list]
print(course_id_list)
assert course_id not in course_id_list
分离 controller 层
web/controller 层：这一层主要负责处理用户请求，进行访问控制、基本参数校验，以及执行不太复杂的业务逻辑。通常使用 Web 框架（如 Flask、Django）处理 HTTP 请求。
对于复杂的参数校验或者代码都应在 service 层进行处理。
# controller/course_controller.py
import logging
from flask import request, Blueprint
from pydantic import ValidationError
from platform_bankend.src.dto.course_dto import CourseDto
from platform_bankend.src.service.course_service import CourseService
# 在调用的时候指定service实现类
course_service: CourseService = CourseService()
# 定义蓝图
course_router = Blueprint("course", __name__, url_prefix="/course")
@course_router.post("")
def create_course():
# 获取 json 格式请求体
course_info = request.get_json()
# 校验请求数据
try:
# 检验参数是否正常
CourseDto.model_validate(course_info)
# 新增课程
course_id = course_service.create(course_info)
return {
"errcode": 0,
"errmsg": "创建课程成功",
"datas": {
"id": course_id
}
}
except ValidationError as ve:
logging.error(ve.errors())
return {
"errcode": 10002,
# 通过 e.errors() 获取错误列表
# repr 为内置函数，返回一个对象的 string 格式
"errmsg": "数据不符合规则"
}
except Exception as e:
logging.error(e)
return {
"errcode": 10009,
"errmsg": "服务出现错误"
}
@course_router.get("/<int:course_id>")
def get_course(course_id):
try:
course = course_service.get(course_id)
if course is None:
return {
"errcode": 10001,
"errmsg": "课程不存在"
}
# 通过 pydantic 把查询结果对象转为字典格式
datas = CourseDto.model_validate(course).model_dump()
return {
"errcode": 0,
"errmsg": "获取课程信息成功",
"datas": datas
}
except Exception as e:
logging.error(e)
return {
"errcode": 10009,
"errmsg": "服务出现错误"
}
@course_router.get("/list")
def list_course():
try:
course_list = course_service.list()
# 使用列表推导式获取全部数据列表，列表中元素为每个查询结果的字典格式
datas = [CourseDto.model_validate(course).model_dump() for course in course_list]
return {
"errcode": 0,
"errmsg": "获取全部课程列表成功",
"datas": datas
}
except Exception as e:
logging.error(e)
return {
"errcode": 10009,
"errmsg": "服务出现错误"
}
@course_router.put("")
def update_course():
# 获取 json 格式请求体
course_info = request.get_json()
# 校验请求数据
try:
# 检验参数是否正常
CourseDto.model_validate(course_info)
# 更新课程
course_service.update(course_info)
return {
"errcode": 0,
"errmsg": "创建更新成功"
}
except ValidationError as ve:
logging.error(ve.errors())
return {
"errcode": 10002,
# 通过 e.errors() 获取错误列表
# repr 为内置函数，返回一个对象的 string 格式
"errmsg": "数据不符合规则"
}
except Exception as e:
logging.error(e)
return {
"errcode": 10009,
"errmsg": "服务出现错误"
}
@course_router.delete("/<int:course_id>")
def delete_course(course_id):
try:
course = course_service.get(course_id)
if course is None:
return {
"errcode": 10001,
"errmsg": "课程不存在"
}
course_id = course_service.delete(course_id)
return {
"errcode": 0,
"errmsg": "课程删除成功",
"datas": {
"id": course_id
}
}
except Exception as e:
logging.error(e)
return {
"errcode": 10009,
"errmsg": "服务出现错误"
}
启动入口
# server.py
from flask import Flask
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase, Session
class Base(DeclarativeBase):
pass
# 配置连接信息
engine = create_engine("mysql+pymysql://root:hogwarts@localhost/course")
# 创建数据库表
Base.metadata.create_all(engine)
# 创建session对象
DBSession = sessionmaker(bind=engine)
db_session: Session = DBSession()
app = Flask(__name__)
if __name__ == '__main__':
# 注册蓝图
from platform_bankend.src.controller.course_controller import course_router
app.register_blueprint(course_router)
app.run(port=5055, debug=True)
启动服务，完成接口测试。
这样对于课程增删改查的管理接口就定义成功了。
# tests/test_course_controller.py
import requests
class TestCourseController:
def setup_class(self):
self.base_url = "http://127.0.0.1:5055/course"
self.course_id = 111
def test_create_course(self):
data = {
"id": self.course_id,
"name": "Python",
"detail": "测试开发"
}
r = requests.post(self.base_url, json=data)
print(r.text)
assert r.status_code == 200
assert r.json().get("errcode") == 0
def test_get_course(self):
get_url = f"{self.base_url}/{self.course_id}"
r = requests.get(get_url)
print(r.text)
assert r.status_code == 200
assert r.json().get("errcode") == 0
assert r.json().get("datas").get("id") == self.course_id
def test_list_course(self):
list_url = f"{self.base_url}/list"
r = requests.get(list_url)
print(r.text)
assert r.status_code == 200
assert r.json().get("errcode") == 0
def test_update_course(self):
data = {
"id": self.course_id,
"name": "Python-更新",
"detail": "测试开发-更新"
}
r = requests.put(self.base_url, json=data)
print(r.text)
assert r.status_code == 200
assert r.json().get("errcode") == 0
def test_delete_course(self):
delete_url = f"{self.base_url}/{self.course_id}"
r = requests.delete(delete_url)
print(r.text)
assert r.status_code == 200
assert r.json().get("errcode") == 0
总结
虽然分层设计会降低代码的耦合提高可测、可拓展性，但分层设计会不可避免的会提高代码的复杂度，所以在实际工作中应结合具体业务、代码复杂度、后期维护等角度进行综合考虑，并通过设计评审后确定最后的项目规范与设计。