严格 DDD 四层架构领域层、应用层、基础设施层、接口层Django 5 DRF成熟生态 灵活 APIView事件驱动领域事件 进程内 Publisher可替换为 Kafka/RocketMQ双数据库支持用户库 订单库独立配置默认 MySQL PostgreSQL统一响应格式{ code, message, data } 全局异常处理器领域层零 Django 依赖纯 Python dataclass方便单元测试自动 OpenAPIdrf-spectacular 生成 Swagger UI零警告零错误Docker 一键起停MySQL PostgreSQL 应用整套编排技术栈技术版本说明Python3.10语言版本使用str | None等新语法Django5.1Web 框架DRF3.15REST 接口层drf-spectacular0.27OpenAPI/Swagger 生成MySQL8.0用户库默认PostgreSQL13订单库默认YAML-配置文件格式工程结构DDD 四层架构DDD 四层架构依赖接口实现接口接口层 InterfacesDRF APIView · URLConf · Serializer应用层 Application用例编排 · 事务边界 · Command/DTO领域层 Domain聚合根 · 业务规则 · 领域事件基础设施层 InfrastructureDjango ORM · 事件发布器 · 数据库重点外层只依赖内层基础设施通过实现领域层定义的接口Repository、EventPublisher完成对内供给保证领域层零框架依赖。工程结构图基础设施层 Infrastructure领域层 Domain应用层 Application接口层 Interfaces发布事件实现产生DRF APIViewSerializer 请求校验URLConf 路由应用服务Command / DTO聚合根 User · OrderRepository 接口领域事件Django ORM ModelRepository 实现EventPublisher 实现MySQLfrogPostgreSQLseed工程结构列表django-ddd/├── config/│ ├── config.yaml # 本地配置与 gin-ddd 字段一致│ └── config.docker.yaml # docker compose 专用├── docs/│ ├── init.mysql.sql # 用户库 DDL 样例数据│ └── init.postgres.sql # 订单库 DDL 样例数据├── src/│ ├── project/ # Django project│ │ ├── settings.py # 读 YAML → DATABASES 等│ │ ├── urls.py # 根路由健康检查 /api 文档│ │ ├── routers.py # app_label → database 路由器│ │ ├── wsgi.py / asgi.py│ │ └── __init__.py│ ├── shared/ # 共享层│ │ ├── apps.py # 启动时构建事件发布器│ │ ├── domain/│ │ │ ├── events.py # DomainEvent 基类│ │ │ └── publisher.py # EventPublisher 抽象│ │ └── infrastructure/│ │ ├── config.py # YAML → dataclass│ │ ├── response.py # 统一响应 ApiResponseSerializer│ │ ├── exceptions.py # 领域异常 全局 handler│ │ └── events.py # InMemoryEventPublisher│ ├── user/ # 用户 bounded context│ │ ├── domain/{user,repository,events}.py # 聚合根/接口/事件│ │ ├── application/{dto,service}.py # Command 应用服务│ │ ├── infrastructure/{models,repository,listeners}.py # ORM 仓储实现 订阅者│ │ ├── interfaces/{serializers,views,urls}.py # DRF 视图 路由│ │ ├── models.py # re-export 供 Django 发现│ │ └── apps.py # AppConfig.ready() 注册监听器│ └── order/ # 订单 bounded context含状态机│ └── ... 同 user 结构├── Dockerfile├── compose.yaml├── manage.py├── pyproject.toml└── requirements.txt各层职责说明层级位置职责关键原则领域层*/domain/聚合根、值对象、领域事件、仓储接口绝不 import django业务规则内聚应用层*/application/用例编排Command/DTO事务边界薄而清晰不实现业务规则基础设施层*/infrastructure/ORM 模型、仓储实现、事件发布器实现领域接口技术细节下沉接口层*/interfaces/DRF Serializer、APIView、URL只处理 HTTP 交互不含业务规则请求生命周期以创建订单为例展示一条 HTTP 请求在四层之间的流转POST /api/orders 完整链路客户端发送 JSON接口层CreateOrderRequest 校验OrderListView.post转换为 Command应用层OrderApplicationService.create应用层检查用户是否存在领域层Order.create() 校验 业务规则基础设施层DjangoOrderRepository.savePostgreSQLINSERT orders基础设施层InMemoryEventPublisher.publishOrderCreatedEvent触发监听器接口层api_response 包装{code, message, data}关键点跨越层次由内层接口控制应用层调用OrderRepository、EventPublisher接口具体实现由基础设施在运行时注入。业务规则集中在聚合根Order.create()内部完成金额合法性、订单号生成等校验应用层不再写 if/else。事件与主流程解耦publish失败只写日志不阻塞请求响应。快速开始1. 环境准备Python 3.10项目使用了str | None等 PEP 604 语法MySQL 8.0 与 PostgreSQL 13或自行选择其一Docker可选用于一键起停2. 初始化数据库默认配置使用双数据库用户库MySQLfrog订单库PostgreSQLseedDjango 的 ORM Meta 配置为managed False不会自动改表结构表结构完全由docs/init.*.sql负责行为和 gin-ddd / nestjs-ddd 对齐。MySQL 用户库直接使用docs/init.mysql.sqlCREATE DATABASE IF NOT EXISTS frog CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;USE frog;CREATE TABLE users (id BIGINT AUTO_INCREMENT PRIMARY KEY,name VARCHAR(50) NOT NULL UNIQUE,email VARCHAR(100) NOT NULL UNIQUE,phone VARCHAR(20),address VARCHAR(255),created_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,updated_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP);PostgreSQL 订单库与当前订单仓储字段一致CREATE DATABASE seed;\c seed;CREATE TABLE orders (id BIGSERIAL PRIMARY KEY,order_no VARCHAR(50) NOT NULL UNIQUE,user_id BIGINT NOT NULL,total_amount DECIMAL(12, 2) NOT NULL DEFAULT 0.00,status VARCHAR(20) NOT NULL DEFAULT PENDING,created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP);数据库适配注意若只用一个数据库在src/project/routers.py中把APP_DB_MAPPING全部指向default即可若订单库改为 MySQL字段BIGSERIAL / TIMESTAMP需替换为BIGINT AUTO_INCREMENT / TIMESTAMP3. 配置应用编辑config/config.yaml与 gin-ddd 字段一致server:host: 0.0.0.0port: 8080debug: truedatabase:user:engine: mysqlhost: localhostport: 3306username: rootpassword: your_passwordname: frogorder:engine: postgreshost: localhostport: 5432username: postgrespassword: your_passwordname: seedlogger:level: INFOevent:publisher: memory # memory / kafka / rocketmq预留说明环境变量CONFIG_FILE可覆盖默认配置路径Docker 里指向config/config.docker.yamlevent.publisher: memory使用进程内发布器切换为 Kafka/RocketMQ 只需在configure_publisher()里加分支4. 安装依赖cd practice-projects/django-dddpython -m venv .venvsource .venv/bin/activate # Windows: .venv\Scripts\activatepip install -r requirements.txtmysqlclient依赖本地 libmysqlclient如果装不上可换成PyMySQL并在manage.py里加pymysql.install_as_MySQLdb()。5. 启动应用# 本地开发python manage.py runserver 0.0.0.0:8080# 生产部署gunicorn --chdir src project.wsgi:application --bind 0.0.0.0:8080 --workers 3Docker 一键起停docker compose up -d --build # 起 MySQL PostgreSQL Appdocker compose logs -f app # 查看日志docker compose down -v # 停止并清理数据卷6. 验证接口curl http://localhost:8080/healthcurl http://localhost:8080/api/userscurl http://localhost:8080/api/ordersAPI 文档Swagger UIhttp://localhost:8080/api/docsOpenAPI Schemahttp://localhost:8080/api/schema如何基于脚手架开发新功能示例新增商品管理模块product。步骤 1新增领域聚合src/product/domain/product.pyfrom __future__ import annotationsfrom dataclasses import dataclass, fieldfrom datetime import datetimedataclassclass Product:name: strprice: floatstock: int 0id: int | None Nonecreated_at: datetime field(default_factorydatetime.utcnow)updated_at: datetime field(default_factorydatetime.utcnow)classmethoddef create(cls, name: str, price: float, stock: int) - Product:if not name:raise ValueError(商品名不能为空)if price 0:raise ValueError(价格必须大于 0)return cls(namename, priceprice, stockstock)步骤 2新增仓储接口src/product/domain/repository.pyfrom abc import ABC, abstractmethodfrom product.domain.product import Productclass ProductRepository(ABC):abstractmethoddef save(self, p: Product) - Product: ...abstractmethoddef find_by_id(self, pid: int) - Product | None: ...abstractmethoddef list_all(self, offset: int, limit: int) - tuple[list[Product], int]: ...步骤 3新增 ORM 模型与仓储实现src/product/infrastructure/# models.pyfrom django.db import modelsclass ProductModel(models.Model):name models.CharField(max_length100, uniqueTrue)price models.DecimalField(max_digits10, decimal_places2)stock models.IntegerField(default0)created_at models.DateTimeField(auto_now_addTrue)updated_at models.DateTimeField(auto_nowTrue)class Meta:app_label productdb_table productsmanaged False # 表结构由 docs/init.*.sql 管理# repository.pyfrom product.domain.product import Productfrom product.domain.repository import ProductRepositoryfrom product.infrastructure.models import ProductModelclass DjangoProductRepository(ProductRepository):def save(self, p: Product) - Product:if p.id is None:row ProductModel.objects.create(namep.name, pricep.price, stockp.stock)else:row ProductModel.objects.get(pkp.id)row.name, row.price, row.stock p.name, p.price, p.stockrow.save()return _to_domain(row)# ...步骤 4新增应用服务src/product/application/service.pyfrom product.application.dto import CreateProductCommand, ProductDTOfrom product.domain.product import Productfrom product.domain.repository import ProductRepositoryclass ProductApplicationService:def __init__(self, repository: ProductRepository):self._repo repositorydef create(self, cmd: CreateProductCommand) - ProductDTO:p Product.create(cmd.name, cmd.price, cmd.stock)p self._repo.save(p)return ProductDTO.of(p)步骤 5新增 HTTP 入口src/product/interfaces/serializers.py用于请求字段校验、views.py写 APIView、urls.py暴露路由再在src/project/urls.py中include(product.interfaces.urls)。步骤 6注册 AppConfig# src/product/apps.pyfrom django.apps import AppConfigclass ProductConfig(AppConfig):name productlabel productdefault_auto_field django.db.models.BigAutoField在src/project/settings.py::INSTALLED_APPS中追加product.apps.ProductConfig。步骤 7新增数据库表CREATE TABLE IF NOT EXISTS products (id BIGINT AUTO_INCREMENT PRIMARY KEY,name VARCHAR(100) NOT NULL UNIQUE,price DECIMAL(10, 2) NOT NULL,stock INT NOT NULL DEFAULT 0,created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMPON UPDATE CURRENT_TIMESTAMP);事件驱动与领域事件事件类型订单事件order.createdorder.paidorder.shippedorder.deliveredorder.cancelledorder.refunded用户事件user.createduser.email.updateduser.phone.updateduser.deleted消息流转领域层只定义事件接口业务代码只和EventPublisher打交道实现可以在内存总线 / Kafka / RocketMQ / RabbitMQ 之间随意切换上层零改动。事件处理发布器实现可替换事件抽象事件产生publish可切换OrderApplicationServiceUserApplicationServiceEventPublisher 接口InMemoryEventPublisher默认 · 进程内KafkaEventPublisherRocketMQ / RabbitMQuser.infrastructure.listeners日志 · 审计 · 通知order.infrastructure.listeners发货 · 下游同步外部消费者服务订单状态机订单的所有合法转移封装在Order聚合根内部外部只能通过pay()/ship()/deliver()/cancel()/refund()等方法触发任何非法转移都会抛DomainError订单状态机paycancelshiprefunddeliverrefundPENDING待支付PAID已支付CANCELLED已取消SHIPPED已发货REFUNDED已退款DELIVERED已送达事件发布与消费关键点发布器抽象shared/domain/publisher.py :: EventPublisher默认InMemoryEventPublisher装配时机SharedConfig.ready()构建发布器各 BC 的AppConfig.ready()注册监听器BC 自治user只订阅自己的事件、order只订阅自己的shared 层永不知道业务事件类型发布失败不影响主流程发布器内部 try/except仅记录日志替换中间件写一个KafkaEventPublisher(EventPublisher)在configure_publisher(kind)里加一个分支即可上层代码零改动# src/shared/apps.pyclass SharedConfig(AppConfig):name shareddef ready(self):kind getattr(settings, DDD_EVENT_PUBLISHER_KIND, memory)_events.configure_publisher(kind)# src/user/apps.pyclass UserConfig(AppConfig):name userdef ready(self):from shared.infrastructure.events import get_publisherfrom user.infrastructure.listeners import register_listenersregister_listeners(get_publisher())INSTALLED_APPS的顺序保证shared → user → order发布器先于订阅者初始化。统一响应与全局异常所有成功响应都经api_response(data, message, code, http_status){ code: 0, message: success, data: { id: 1, name: jarry } }领域层抛DomainError / NotFoundError / ValidationErrorDRF 全局异常处理器shared.infrastructure.exceptions.global_exception_handler统一包装成相同结构避免裸异常栈泄露到响应。# settings.pyREST_FRAMEWORK {EXCEPTION_HANDLER: shared.infrastructure.exceptions.global_exception_handler,}多数据库与路由器# src/project/routers.pyAPP_DB_MAPPING {user: default, # MySQL (frog)order: order_db, # PostgreSQL (seed)}class AppLabelRouter:def db_for_read(self, model, **hints):return APP_DB_MAPPING.get(model._meta.app_label, default)db_for_write db_for_readdef allow_migrate(self, db, app_label, **hints):return db APP_DB_MAPPING.get(app_label, default)通过app_label自动路由业务代码感知不到多数据库的存在。新增第三个库时只需在APP_DB_MAPPING里加一行不用改业务。开发规范命名建议领域聚合名词如Order、User、Product应用服务XxxApplicationService仓储接口XxxRepository仓储实现DjangoXxxRepository视图XxxViewAPIView 子类命令对象XxxCommandapplication/dto.py分层原则领域层不依赖 Django、不依赖 DRF、不依赖任何 ORM应用层只做编排与事务协调不写 if/else 业务判断基础设施只提供技术实现不暴露 ORM 对象给上层接口层只负责 HTTP 交互Serializer 只做格式校验跨上下文原则user与order互不 import跨 BC 的只读组合视图放在project/urls.pyshared层不 import 任何 BC订阅者都放在各自 BC 里常用命令# 依赖管理pip install -r requirements.txtpip freeze requirements.txt# 启动python manage.py runserver 0.0.0.0:8080python manage.py checkpython manage.py spectacular --file docs/openapi.yaml# Dockerdocker compose up -d --builddocker compose logs -f appdocker compose down -v# 测试推荐pytest