资讯动态

DDD+Flask+Python实践:构建可维护博客系统的领域模型与分层架构

发布时间:2026/10/20 20:01:22 来源:尧图企业网站定制
写博客系统谁不会呢控制器、模型、模板一套下来增删改查就完事了。我当时接到这个项目时也是这么想的一个博客系统最典型不过的 CRUD无非是文章、分类、评论、标签几个表换个花样。可真动手之后发现需求里光是文章状态流转、评论审核、标签去重和权限校验就够写一阵子。如果继续用那种视图里写查询模型里塞逻辑的写法后期一定糊。后来我把这套东西重构成 DDD Python Flask RESTful 的架构效果比预期好很多。这篇文章就是把我实际落地时的设计思路、代码组织方式、踩过的坑统统写下来给想用领域驱动设计来实现 Web 项目的同学做个参考。这里面向的读者不是刚会 Python 语法的萌新而是已经写过几个 Flask 项目、但对 DDD 怎么和 Flask 配合感到迷茫的人。1. 为什么一个博客系统值得用 DDD1.1 博客系统真的简单吗很多人一听博客系统就觉得是玩具项目但你把需求展开就会发现业务规则其实不少。以文章为例一篇文章有草稿、已发布、已下线、已删除四种状态。从草稿到发布需要走审核已发布的文章不能直接被删除只能先下线再进回收站回收站里的文章保留 30 天后才能物理清除。评论呢用户评论要审核管理员审核通过才显示作者可以置顶某条评论还可以屏蔽某个用户的所有评论。再加上标签的合并、文章归档、按热度排序这些规则如果全部堆在视图函数里后期的维护成本会非常恐怖。我在重构前就试过所有逻辑都写在 Flask 的蓝图Blueprint里一个article_create函数处理文章创建、标签联动、状态初始化和权限校验代码量直接奔着 200 行去了。更麻烦的是当你需要从另外一个接口复用创建文章这个业务动作时你只能复制粘贴改起来到处找。这就是选择 DDD 的直接原因博客系统的业务逻辑并不简单尤其是在长期迭代的场景下更需要把业务规则从 Web 框架里剥离出来。DDD 的核心思想就是让领域层成为系统的内核Flask、SQLAlchemy、RESTful 这些统统只是外围设施。1.2 DDD 分层到底分哪些层DDD 经典的四层架构我实际落地时简化了一下但也保持了核心边界分层职责依赖方向接口层处理 HTTP 请求和响应参数校验返回状态码依赖应用层应用层编排领域对象完成一个用例启动事务不写业务规则依赖领域层领域层实体、值对象、聚合根、领域服务、仓储接口、领域异常不依赖任何框架基础设施层实现仓储接口、ORM 映射、数据库连接、第三方 SDK依赖领域层接口最常见的误区是把接口层和应用层混为一谈。我见过很多项目把 Controller 里直接写业务规则比如请求进来后先查数据库再判断状态再更新。在 DDD 里这个动作应该拆成接口层只做参数转换和鉴权然后调用应用层的一个方法例如publish_article(article_id, operator_id)应用层内部调用领域服务完成状态校验和状态变更最后通过仓储接口持久化。分层带来的好处是依赖方向单向向下领域层不 import 任何flask或sqlalchemy。我当时验证分层是否干净的方法很简单先把flask、sqlalchemy从 requirements 里临时去掉然后看看领域层还能不能 import 成功。如果能说明分层基本合格。有必要说明一点我是先明确了需求边界才决定引入 DDD。如果你只是写个一次性报表页面或者项目周期特别短、业务几乎不变DDD 确实是过度设计。博客系统这种业务规则多的场景才值得这么做。2. 领域建模先把文章和评论的关系理清2.1 聚合与聚合根怎么划分领域建模是 DDD 落地最重要也最容易卡壳的一步。我当时卡了很久的一个问题是文章、评论、标签、分类到底谁是聚合根它们之间怎么关联。最终我采用的是文章Article是聚合根评论Comment是文章聚合内的实体标签Tag是值对象。为什么这么划分可以从一致性来考虑。评论不能独立存在它必须依附于某篇文章删除文章时评论必须一并处理。这就是聚合边界存在的意义聚合内部的对象要保证一致性。而标签更像是文章的一个属性集合它本身没有独立生命周期标签重命名时不需要依赖文章所以作为值对象由文章聚合持有。这里有个容易误解的点DDD 里的对象并不等同于数据库表。很多人听到评论是实体就把数据库里的comments表对应成一个实体类然后所有 GET 接口都从这个表里直接查。但我的做法是评论的查询如果出于展示目的完全可以走独立的查询模型只有涉及修改评论状态、删除评论时才通过Article聚合来操作。这个思路对应 CQRS虽然我在这套博客系统里没有完全引入 CQRS但这个原则帮我避免了大量为了查数据而绕一大圈的尴尬。聚合根的标识是id实体的标识可以是id也可以是一组业务属性组合。比如文章里的评论我用comment_id作为实体标识而标签我用属性组合(name, slug)来标识这样标签天然去重。2.2 值对象、实体和领域服务领域建模时我把文章定义成这样的 Python 数据类from dataclasses import dataclass, field from datetime import datetime from typing import List, Optional from domain.models.comment import Comment from domain.models.tag import Tag from domain.exceptions import DomainError dataclass class Article: id: Optional[int] title: str content: str slug: str status: str # draft / pending / published / offline / deleted tags: List[Tag] field(default_factorylist) comments: List[Comment] field(default_factorylist) created_at: datetime field(default_factorydatetime.utcnow) published_at: Optional[datetime] None def publish(self, operator_id: int) - None: if self.status ! draft: raise DomainError(只有草稿状态的文章才能提交发布) if not self.title.strip() or not self.content.strip(): raise DomainError(标题和内容不能为空) self.status pending self.published_at datetime.utcnow() def approve(self, operator_id: int) - None: if self.status ! pending: raise DomainError(只有待审核的文章才能通过审核) self.status published注意我把publish和approve这种状态变更方法直接放在领域对象上而不是放在 Service 里。这才符合 DDD 的领域行为原则。之前的 MVC 写法里这些逻辑被放在 Service 类中Service 里先查出来再判断再赋值模型变成纯粹的容器业务规则散落各处。但是有一些行为不属于某个具体实体而是协调多个聚合或外部服务我把它提取成领域服务。比如发布文章并发送通知就不适合放在Article上因为涉及通知渠道我定义了一个ArticlePublishServiceclass ArticlePublishService: def __init__(self, article_repo, notifier): self.article_repo article_repo self.notifier notifier def publish(self, article_id: int, operator_id: int) - None: article self.article_repo.find_by_id(article_id) if article is None: raise DomainError(文章不存在) article.publish(operator_id) self.article_repo.save(article) self.notifier.notify_article_published(article)领域服务和应用层的区别在于领域服务内仍然在操作领域对象、维持业务不变量应用服务则只是协调动作不包含业务判断。2.3 仓储接口设计仓储Repository是领域层和数据访问的桥梁。在 DDD 中仓储接口定义在领域层实现放在基础设施层。这样领域层依赖的是抽象而不是 SQLAlchemy 的 Session 或某个具体数据库。我当时定义了下面几个核心仓储接口from abc import ABC, abstractmethod from typing import List, Optional from domain.models.article import Article class ArticleRepository(ABC): abstractmethod def find_by_id(self, article_id: int) - Optional[Article]: 按 ID 查找文章未找到时返回 None abstractmethod def save(self, article: Article) - None: 保存或更新文章 abstractmethod def delete(self, article_id: int) - None: 物理删除文章 abstractmethod def find_published(self, page: int, size: int) - List[Article]: 分页查询已发布文章为什么仓储的方法如此粗粒度我记得之前写 DAO 的时候喜欢拆一堆update_title、update_content、update_status之类的方法结果每一次调用都是一次 UPDATE逻辑一复杂就出现并发问题。现在仓储只提供save方法领域对象在内存中完成状态变更再由仓储整体持久化。这个模式更好保持聚合一致性。当然仓储接口不是一成不变的。比如查看文章列表时为了性能需要避免把评论都查出来我后来又增加了查询方法但保持返回领域对象不直接暴露 ORM 实体。这一点是 DDD 和其他风格比较大的区别。3. 在 Flask 中落地 DDD目录结构与路由实现3.1 项目目录到底怎么摆落地 DDD 时最直接的问题是目录结构。我之前见过一些项目标榜 DDD实际上只是把 Controller 换成 Service目录仍然是models、views、utils换汤不换药。参考了我自己的实践最终目录结构长这样blog_system/ ├── app.py # Flask 入口创建应用 ├── config.py # 配置 ├── domain/ # 领域层 │ ├── __init__.py │ ├── models/ │ │ ├── __init__.py │ │ ├── article.py │ │ ├── comment.py │ │ └── tag.py │ ├── repositories/ │ │ ├── __init__.py │ │ ├── article_repo.py │ │ └── comment_repo.py │ ├── services/ │ │ ├── __init__.py │ │ ├── article_publish_service.py │ │ └── comment_review_service.py │ └── exceptions.py ├── application/ # 应用层 │ ├── __init__.py │ ├── article_app.py # 文章用例编排 │ └── comment_app.py # 评论用例编排 ├── infrastructure/ # 基础设施层 │ ├── __init__.py │ ├── db.py # SQLAlchemy 初始化 │ ├── orm/ │ │ ├── __init__.py │ │ ├── article_orm.py │ │ ├── comment_orm.py │ │ └── tag_orm.py │ └── repositories/ │ ├── __init__.py │ ├── sql_article_repo.py │ └── sql_comment_repo.py ├── interfaces/ # 接口层 │ ├── __init__.py │ ├── api/ │ │ ├── __init__.py │ │ ├── article_api.py │ │ ├── comment_api.py │ │ └── tag_api.py │ └── schemas.py └── tests/ ├── __init__.py ├── domain/ ├── application/ └── infrastructure/关键在于domain下的代码不 importinfrastructure和interfaces的东西。审查代码时我当时用了一个笨办法在domain目录里全局搜索from flask和from sqlalchemy如果搜到任何一行就说明架构被破坏了。如果你觉得这套目录太繁琐也可以做更轻量的折叠把 ORM 映射直接合并到具体仓储里但接口抽象、依赖方向这些原则不要妥协。3.2 用蓝图组织 RESTful 接口Flask 的蓝图天然适合做 RESTful 接口的模块划分。我按资源来组织蓝图而不是按操作来组织。比如文章相关的接口全部挂在article_bp下from flask import Blueprint article_bp Blueprint(article, __name__, url_prefix/api/articles) article_bp.route(, methods[GET]) def list_articles(): ... article_bp.route(/int:article_id, methods[GET]) def get_article(article_id: int): ... article_bp.route(, methods[POST]) def create_article(): ... article_bp.route(/int:article_id, methods[PUT]) def update_article(article_id: int): ... article_bp.route(/int:article_id/publish, methods[POST]) def publish_article(article_id: int): ... article_bp.route(/int:article_id/comments, methods[GET]) def list_article_comments(article_id: int): ...这里有个很小的设计点/api/articles用复数名词路径里不出现动词publish动作通过子资源/publish来呈现。刚开始我设计过/api/article/publish?id1这样的接口后来测试时严格按照 RESTful 语义去审发现既不符合资源命名规范也不容易扩展就改成了现在这样。关于状态码接口层需要区分几种情况创建成功返回201 Created参数不合法返回400 Bad Request资源不存在返回404 Not Found权限不足返回403 Forbidden业务规则不满足返回422 Unprocessable Entity。我在这个项目里统一用一个错误处理器捕获领域异常并映射到 HTTP 状态码。from flask import jsonify from domain.exceptions import DomainError from infrastructure.db import DatabaseError def register_error_handlers(app): app.errorhandler(DomainError) def handle_domain_error(error): return jsonify({code: domain_error, message: str(error)}), 422 app.errorhandler(404) def handle_not_found(error): return jsonify({code: not_found, message: 资源不存在}), 404在实际写的时候不要把DomainError直接暴露给前端。接口层要做一层包装返回给前端的是稳定的错误码和错误结构而不是一段 Python 异常的堆栈信息。3.3 请求参数到领域对象的转换写 RESTful 接口时最烦的就是参数校验。Flask 本身不提供强校验我试过在视图函数里一点点手写if title not in data代码很快就膨胀了。后来我引入了一个轻量方案用 Pydantic 或手动 dataclass 定义请求 DTO在接口层完成校验然后转换为应用层的输入对象。一个比较典型的创建文章的接口处理流程from dataclasses import dataclass dataclass class CreateArticleRequest: title: str content: str slug: str tags: list def parse_create_article_request(payload: dict) - CreateArticleRequest: title payload.get(title, ).strip() content payload.get(content, ).strip() slug payload.get(slug, ) or slugify(title) tags payload.get(tags, []) if not title: raise InvalidRequest(标题不能为空) if len(title) 100: raise InvalidRequest(标题长度不能超过100个字符) return CreateArticleRequest(titletitle, contentcontent, slugslug, tagstags) article_bp.route(, methods[POST]) def create_article(): payload request.get_json(forceTrue) req parse_create_article_request(payload) article_id article_app.create_article( titlereq.title, contentreq.content, slugreq.slug, tagsreq.tags, operator_idrequest.current_user.id, ) return jsonify({id: article_id, status: created}), 201有人会问为什么不直接把领域对象放进 Flask 视图里我这里给出的答案是领域对象是业务内核不应该暴露给 HTTP 层。如果接口层直接操作Article对象你很难阻止某个视图里无意间修改了领域对象的状态而没有走仓储持久化。DTO 的引入相当于一层隔离让接口只做消息转换。在参数里我没有直接传user_id而是通过某种方式从登录态中取出当前用户。这是很常见的做法当前登录用户往往是鉴权中间件注入的不应该信任请求体的字段。3.4 依赖注入的轻量实现Flask 本身没有内置依赖注入容器而 DDD 落地时各层之间又需要解耦。我的做法是写一个非常小的手动容器不用 Flask-SQLAlchemy 插件自动把模型注册到全局。class Container: def __init__(self): self._services {} def register(self, name: str, factory): self._services[name] factory def resolve(self, name: str): factory self._services[name] if isinstance(factory, type): return factory() return factory() container Container() def create_app(): app Flask(__name__) db.init_engine(app.config[DATABASE_URI]) article_repo SqlArticleRepository(db.session_factory) comment_repo SqlCommentRepository(db.session_factory) article_publish_service ArticlePublishService(article_repo, Notifier()) article_app ArticleAppService( article_repoarticle_repo, article_publish_servicearticle_publish_service, ) container.register(article_app, article_app) container.register(comment_app, lambda: CommentAppService(...)) register_blueprints(app) register_error_handlers(app) return app这样的手动依赖注入在小型项目里够用不需要引入额外的库。关键收益是你在测试时可以用假仓储替换真实仓储从而对应用层进行独立测试。如果你觉得手动容器维护麻烦也可以直接在create_app函数里采用闭包方式传递依赖或者在每个蓝图里调用container.resolve(article_app)。但不要把这些依赖写成模块级全局变量否则测试时会互相污染。4. 持久化与事务SQLAlchemy 和仓储的真实配合4.1 SQLAlchemy 映射与 ORM到了基础设施层我遇到一个比较现实的问题领域模型已经有Articledataclass 了SQLAlchemy 的 ORM 模型怎么处理如果直接把 SQLAlchemy 模型当领域模型用那么 DDD 领域层就名存实亡了。我采取的是双模型策略ORM 模型负责表结构映射领域模型负责业务表达两者在仓储里互相转换。ORM 模型大概长这样from datetime import datetime from sqlalchemy import ( Column, Integer, String, Text, DateTime, ForeignKey, Table, ) from sqlalchemy.orm import declarative_base, relationship Base declarative_base() article_tag_association Table( article_tag, Base.metadata, Column(article_id, Integer, ForeignKey(article.id), primary_keyTrue), Column(tag_id, Integer, ForeignKey(tag.id), primary_keyTrue), ) class ArticleORM(Base): __tablename__ article id Column(Integer, primary_keyTrue) title Column(String(100), nullableFalse) content Column(Text, nullableFalse) slug Column(String(120), uniqueTrue, nullableFalse) status Column(String(20), nullableFalse, defaultdraft) created_at Column(DateTime, defaultdatetime.utcnow) published_at Column(DateTime, nullableTrue) tags relationship(TagORM, secondaryarticle_tag_association) comments relationship(CommentORM, lazyselect)对应仓储实现中关键一步是从 ORM 转换为领域对象class SqlArticleRepository(ArticleRepository): def __init__(self, session_factory): self.session_factory session_factory def find_by_id(self, article_id: int) - Optional[Article]: session self.session_factory() orm_article session.query(ArticleORM).get(article_id) if orm_article is None: return None article self._to_domain(orm_article) return article staticmethod def _to_domain(orm_article: ArticleORM) - Article: tags [Tag(namet.name, slugt.slug) for t in orm_article.tags] comments [Comment(...) for c in orm_article.comments] return Article( idorm_article.id, titleorm_article.title, contentorm_article.content, slugorm_article.slug, statusorm_article.status, tagstags, commentscomments, created_atorm_article.created_at, published_atorm_article.published_at, ) def save(self, article: Article) - None: session self.session_factory() orm_article session.query(ArticleORM).get(article.id) if orm_article is None: orm_article ArticleORM() session.add(orm_article) self._apply_to_orm(orm_article, article) session.commit()这一步代码虽多但价值很大领域层不依赖 SQLAlchemy你可以随时把仓储实现换成 MongoDB 或内存库只要接口不变应用层和领域层完全不用动。有一个影响性能的细节find_by_id时把评论都查出来可能导致某篇文章评论特别多时查询很慢。我后来在接口层做了一个区分查看文章详情时返回完整内容其中评论是分页的而编辑文章时不需要评论。于是我在仓储里增加了find_without_comments方法避免无关数据全部加载。4.2 事务边界应该画在哪事务边界是 DDD 项目里最容易出现问题的点。我一开始把session.commit()写在仓储里后来发现应用层想同时操作文章和评论时会产生多个写入提交一旦后者失败前者已经提交了数据就处于不一致状态。正确做法是应用层控制事务仓储只负责在会话内读写不负责提交和回滚。我使用 Flask-SQLAlchemy 的db.session时会在应用服务里显式管理事务边界。from infrastructure.db import db class ArticleAppService: def __init__(self, article_repo, article_publish_service): self.article_repo article_repo self.article_publish_service article_publish_service def publish_article(self, article_id: int, operator_id: int) - None: try: with db.session.begin(): self.article_publish_service.publish(article_id, operator_id) except Exception: db.session.rollback() raise这里我尽量不依赖 Flask-SQLAlchemy 的自动提交因为自动提交会在请求结束时统一处理可一旦业务复杂到需要跨服务协调你根本看不清楚事务到底在哪里闭合。显式开启事务边界虽然多写几行但可读性和可控性都大幅提升。事务边界和锁也有关系。如果两个管理员同时对同一篇文章执行审核一个通过一个驳回最后状态取决于后提交的人这在业务上是不允许的。我的做法是在文章表上加上版本号用乐观锁控制状态变更。DDD 里这种做法对应的是给聚合根一个版本号仓储保存时校验版本。# Article 领域对象增加 version 字段 dataclass class Article: ... version: int 0 # 仓库保存时校验版本更新版本号 def save(self, article: Article) - None: ... orm_article.version article.version 1刚开始可能觉得博客系统哪需要什么乐观锁等到真的出现并发状态变更问题就晚了。当然如果你明确知道这个系统只是个人博客并发几乎为零那也可以不引入但至少要清楚这个取舍的原因。5. 实际踩过的坑与排查实录5.1 循环依赖把接口从实现中抽出来我最早写的时候在领域层直接 import 了 SQLAlchemy 模型导致领域层报错。今天重新复盘发现这是 DDD 新手几乎都会踩的坑你定义一个ArticleRepository接口同时又在另一个包里实现一个SqlArticleRepository但领域层某个领域服务想用这个仓储于是你图方便直接在领域服务里from infrastructure.repositories import SqlArticleRepository。这样一来领域层就反向依赖了基础设施层DAG 被破坏后续想替换实现几乎不可能。更麻烦的是Python 的 import 在项目大了以后会出现循环引用问题比如 ORM 模型需要引用领域模型领域模型又通过某种方式引用 ORM程序启动时直接各种ImportError。我的解决原则其实很简单领域层只 import 接口和 dataclass任何具体实现都放在应用层组装时再注入。为了帮助新人审查我当时写了一个自动化检查脚本在 CI 里对domain目录做 AST 解析检测是否出现infrastructure或interfaces的引用发现就报错。5.2 懒加载导致的响应序列化问题在 Flask 里写 RESTful API 时我碰到过一个非常经典的问题ORM 模型有懒加载属性比如ArticleORM的comments是lazyselect当你把 ORM 对象直接扔给jsonify时会因为懒加载对象已经被 session 关闭而报DetachedInstanceError或者因为序列化时触发额外查询导致性能雪崩。这个问题在 DDD 双模型下反而被天然规避了接口层根本没有机会接触到 ORM 模型。当你从应用层拿到Article领域对象后需要序列化为 JSON 时不能直接把 dataclass 的所有字段抛出去要专门为响应定义 DTO。def serialize_article(article: Article) - dict: return { id: article.id, title: article.title, content: article.content, slug: article.slug, status: article.status, created_at: article.created_at.isoformat(), published_at: article.published_at.isoformat() if article.published_at else None, tags: [tag.name for tag in article.tags], }这个 DTO 序列化函数可以放在接口层的schemas.py好处是即使领域模型字段再变对外接口的字段结构仍然稳定可控。我见过太多项目因为直接序列化 ORM 模型导致接口字段一会儿多一个内部字段一会儿又是None。不过这里也要提醒一点分页查询时如果把每篇文章的标签和评论全部序列化N1 查询问题会立刻冒头。我在仓储查询里对列表页做了特殊处理只联表查询需要的字段并通过 selectin 加载标签避免逐条查询。5.3 RESTful 资源命名不一致做 RESTful 接口时我最开始犯过一个很低级的错误有的接口用article_id有的用id有的接口返回的 JSON 里文章用post字段另一些用article。前端对接的人都被搞晕了一直在问这个接口到底返回什么结构。后来我强制自己遵守一套约定并写进了项目的接口文档里资源主路径查询参数/说明文章/api/articles?page1size20statuspublished单篇文章/api/articles/{id}返回文章详情文章的评论/api/articles/{id}/comments返回该文章评论列表标签/api/tags返回标签列表当前用户/api/me返回登录用户数据接口内部响应结构统一为{code: 0, data: ..., message: success}或者直接把业务数据放在顶层不混用。关键是要让团队看到接口路径就能猜到资源关系不要在 URL 里出现getArticle这种行为命名。很多人可能会说 RESTful 并不强制要求这种命名但对一个长期维护的项目来说一致性比风格本身更重要。我当时做接口评审时把接口路径一致性作为进入测试的前置条件否则打回重做。5.4 测试领域层时不想碰数据库DDD 测试和传统 Flask 测试最大的不同是测试领域层和应用层时完全可以把数据库替换掉。因为领域层只依赖仓储接口测试时用一个内存实现或者 Mock 仓储即可。举个例子测试Article的状态流转不需要任何数据库from domain.models.article import Article from domain.exceptions import DomainError import pytest def test_publish_draft_article_to_pending(): article Article(idNone, titlet, contentc, slugt) article.publish(operator_id1) assert article.status pending def test_publish_non_draft_article_should_raise(): article Article(id1, titlet, contentc, slugt, statuspublished) with pytest.raises(DomainError): article.publish(operator_id1)应用层测试时用一个 FakeArticleRepository 代替 SQLAlchemy 实现验证业务流程是否正确class FakeArticleRepository(ArticleRepository): def __init__(self): self._articles {} self._id 1 def find_by_id(self, article_id): return self._articles.get(article_id) def save(self, article): if article.id is None: article.id self._id self._id 1 self._articles[article.id] article def find_published(self, page, size): return [a for a in self._articles.values() if a.status published]这样做的好处是测试速度极快而且不依赖本地数据库环境。只有基础设施层的仓储测试才真正连接 SQLite 或测试数据库验证 SQL 映射是否正确。我当时把整个测试分了三层领域测试跑得飞快应用测试也用 Fake 仓储接口测试用 Flask 的test_client配合内存数据库整体测试很快。6. 后续还能怎么扩展6.1 从当前架构到更复杂的 DDD 实践这套博客系统目前还没有引入事件驱动。其实发布文章、审核评论这种动作非常适合做领域事件比如ArticlePublished,CommentApproved。如果你想让应用层更轻量可以引入事件总线在领域对象状态变更时抛出事件应用层订阅之后再做发送通知、更新缓存等副作用。引入领域事件后聚合根的一个操作可以触发多个外部动作而不需要在应用服务里把代码写成一长串。比如用户发文章后要清缓存、发通知、更新归档这些如果用同步编排代码会非常长。事件驱动之后主流程还是先保存文章事件由事件总线统一派发。另外如果你觉得博客系统里查询越来越复杂比如需要按标签、按日期、按热度同时过滤可以考虑引入查询模型。把文章展示数据单独存成一张宽表用独立查询仓储去读而不是每次都在主聚合上做复杂查询。这就是 CQRS 的简化版也是 DDD 往更深实践走很自然的一步。不过我要说句实在话当项目还没到那个规模时不要提前引入异步消息和事件溯源否则复杂度会非常难控制。6.2 一些个人体会整个项目做下来我最深的一个体会是DDD 不是一件魔法外衣不是说用了这套分层你的代码就干净了。真正难的是领域建模是你能不能把文章状态流转、评论审核这些业务规则准确放进领域对象里。建模不是一次完成的我初版建模和最终版差别非常大初版把Tag设计成实体后来发现博客系统对标签不需要那么强的生命周期管理改成值对象后代码简单了很多。第二个体会是DDD 和 Flask 的配合需要你在工程上做一些折衷。比如我不打算完全引入数据库迁移工具链的额外复杂度就保留了 SQLAlchemy 的自动建表同时让 ORM 模型和领域模型使用不同的命名空间。随着项目变大这种折衷是否合理要定期重新审视不要因为刚开始这么定了就一直不改。最后再分享一个细节如果你要用 DDD请务必从第一个用例开始就坚持分层。最怕是前期图省事在视图里塞逻辑等到后期再想去重构分层改起来非常痛。我踩过几次坑之后发现从新增一个接口的第一行代码起就明确它属于接口层、应用层还是领域层比任何代码评审都管用。

读完文章,也想定制专属网站?

尧图设计师 24 小时内与您沟通定制方案

免费获取报价 →
↑