django-clean-drf

Django Clean DRF 架构师

概述

此技能提供专业指导，帮助你遵循整洁架构原则创建 Django REST Framework 应用。它强制采用严格的分层架构，分离各项职责、最大限度地提高可测试性，并生成可维护的生产级代码。

架构层：

HTTP Layer (Views/Serializers)
    ↓
Application Layer (Use Cases + DTOs)
    ↓
Domain Layer (Services + Models)
    ↓
Infrastructure Layer (ORM, External APIs, Cache)

主要功能：

• 采用分层架构，并遵守严格的依赖规则（依赖只能向内流动）
• 使用包含 dataclass 输入/输出 DTO 的用例模式
• 使用服务层处理领域逻辑（静态查询、实例变更）
• 采用委托给用例的轻量视图
• 分离读取序列化器与创建/更新序列化器
• 显式处理错误，业务逻辑不使用异常
• 使用现代 Python 3.12+ 和 Django 5+ 特性
• 从设计上支持完整类型提示和可测试性

适用场景

遇到以下触发情形时，请调用此技能：

新应用设置：

• “为……创建一个新的 Django API”
• “使用整洁架构启动一个新的 DRF 项目”
• “使用分层架构设置一个新应用”
• “初始化一个领域驱动的 Django 应用”
• “按照 SOLID 原则搭建一个 Django 应用”

用例实现：

• “为……创建一个用例”
• “实现业务逻辑……”
• “添加一个新操作/动作……”
• “以原子方式处理此事务”

服务层：

• “为……创建一个服务”
• “这段业务逻辑应该放在哪里？”
• “实现领域验证……”
• “遵循整洁架构查询数据”

架构问题：

• “如何组织这个 Django 应用？”
• “……应该由哪一层处理？”
• “如何避免臃肿的视图/模型？”
• “如何让代码可测试？”
• “在 Django 中应用 SOLID 原则”

API 设计：

• “为……创建 API 端点”
• “设计序列化器……”
• “为此实体实现 CRUD”

指令

处理 Django Clean DRF 请求时，请遵循以下工作流：

1. 分析需求并建立上下文

了解领域：

• 涉及哪些实体/模型？
• 需要哪些操作/动作？
• 必须强制执行哪些业务规则？
• 是否存在外部依赖（API、队列）？

检查现有项目结构：

• 查看现有应用及其组织方式
• 识别当前使用的模式
• 检查 Django/Python 版本，确认功能是否可用

确定范围：

• 单个用例还是完整 CRUD？
• 新建应用还是扩展现有应用？
• 仅包含 API，还是也包含管理后台？

2. 加载相关参考文档

根据任务参考相应的随附文档：

架构：

• 新应用设置 → references/project-structure.md
• 用例实现 → references/use-cases-pattern.md
• 服务层 → references/services-pattern.md
• 模型和领域逻辑 → references/models-domain.md
• 视图和序列化器 → references/views-serializers.md
• 测试 → references/testing-clean-arch.md
• 完整示例 → references/examples.md

API 开发：

• 查询优化 / N+1 → references/query-optimization.md
• 分页、筛选、搜索 → references/api-patterns.md
• 身份验证 / JWT / 权限 → references/authentication.md
• 生产环境部署 → references/production-api.md
• 用于 API 的 Django Admin → references/django-admin.md

代码质量：

• 编码规范 / Ruff / 安全 → references/coding-standards.md

3. 遵循整洁架构原则进行实现

重要：层间依赖规则

Views → Use Cases → Services → Models
  ↓         ↓           ↓         ↓
Serializers  DTOs    (internal)   ORM

• 依赖只能向内流动（外层依赖内层）
• 绝不能在用例中导入视图，也不能在服务中导入用例

用例模式：

from dataclasses import dataclass
from uuid import UUID
from django.db import transaction

@dataclass(frozen=True, slots=True)
class CreateOrderInput:
    customer_id: UUID
    items: list['OrderItemInput']
    notes: str | None = None

@dataclass(frozen=True, slots=True)
class CreateOrderOutput:
    success: bool
    order_id: UUID | None = None
    error: str | None = None

class CreateOrderUseCase:
    def __init__(
        self,
        order_service: OrderService,
        inventory_service: InventoryService,
    ):
        self._order_service = order_service
        self._inventory_service = inventory_service

    @transaction.atomic
    def execute(self, input_dto: CreateOrderInput) -> CreateOrderOutput:
        # Validate
        is_valid, error = self._inventory_service.validate_availability(input_dto.items)
        if not is_valid:
            return CreateOrderOutput(success=False, error=error)

        # Execute
        order = self._order_service.create(
            customer_id=input_dto.customer_id,
            items=input_dto.items,
        )
        return CreateOrderOutput(success=True, order_id=order.id)

服务模式：

class OrderService:
    # STATIC methods for QUERIES (no state changes)
    @staticmethod
    def get_by_id(order_id: UUID) -> Order | None:
        return Order.objects.select_related('customer').filter(id=order_id).first()

    # INSTANCE methods for MUTATIONS (state changes)
    def create(self, customer_id: UUID, items: list[OrderItemInput]) -> Order:
        order = Order.objects.create(customer_id=customer_id)
        for item in items:
            OrderItem.objects.create(order=order, **item.__dict__)
        return order

    # VALIDATION returns tuple[bool, str]
    def validate_can_cancel(self, order: Order) -> tuple[bool, str]:
        if order.status == Order.Status.SHIPPED:
            return False, "Cannot cancel shipped orders"
        return True, ""

轻量视图模式：

class CreateOrderView(APIView):
    permission_classes = [IsAuthenticated]

    def post(self, request: Request) -> Response:
        serializer = CreateOrderSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)

        use_case = CreateOrderUseCase(
            order_service=OrderService(),
            inventory_service=InventoryService(),
        )
        input_dto = CreateOrderInput(**serializer.validated_data)
        output = use_case.execute(input_dto)

        if not output.success:
            return Response({'error': output.error}, status=400)
        return Response({'order_id': str(output.order_id)}, status=201)

序列化器分离：

# WRITE serializer (input validation)
class CreateOrderSerializer(serializers.Serializer):
    items = OrderItemInputSerializer(many=True, min_length=1)
    notes = serializers.CharField(max_length=500, required=False)

# READ serializer (output display)
class OrderReadSerializer(serializers.ModelSerializer):
    customer_name = serializers.CharField(source='customer.name', read_only=True)

    class Meta:
        model = Order
        fields = ['id', 'customer_name', 'status', 'created_at']
        read_only_fields = fields

包含领域逻辑的模型：

class Order(models.Model):
    class Status(models.TextChoices):
        PENDING = 'pending', 'Pending'
        CONFIRMED = 'confirmed', 'Confirmed'
        SHIPPED = 'shipped', 'Shipped'

    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
    customer = models.ForeignKey('Customer', on_delete=models.PROTECT, related_name='orders')
    status = models.CharField(max_length=20, choices=Status.choices, default=Status.PENDING)
    created_at = models.DateTimeField(auto_now_add=True)

    class Meta:
        ordering = ['-created_at']
        indexes = [models.Index(fields=['customer', 'status'])]

    @property
    def is_cancellable(self) -> bool:
        return self.status in (self.Status.PENDING, self.Status.CONFIRMED)

4. 验证并完成

架构检查清单：

• 视图保持轻量（逻辑少于 20 行）
• 用例处理单一操作
• 服务包含可复用的领域逻辑
• 依赖关系仅向内流动
• 各层之间没有循环导入

代码质量检查清单：

• 所有公共接口都有类型提示
• Dataclass DTO 使用 frozen 和 slots
• 业务逻辑错误不使用异常 → 返回 Output
• 验证返回 tuple[bool, str]
• 状态变更由 @transaction.atomic 包裹

性能检查清单：

• 访问 ForeignKey 时使用 select_related
• 访问反向关联时使用 prefetch_related
• 为频繁查询的字段添加索引
• 序列化器中没有 N+1 查询

目录结构

apps/<app_name>/
├── __init__.py
├── models.py              # Domain entities with @property logic
├── views.py               # Thin HTTP handlers
├── serializers.py         # Read and Write serializers
├── urls.py                # Route definitions
├── admin.py               # Django Admin
├── services/
│   ├── __init__.py        # Export services
│   └── <entity>_service.py
├── use_cases/
│   ├── __init__.py        # Export use cases
│   └── <action>_<entity>.py  # One file per use case
├── tests/
│   ├── __init__.py
│   ├── test_models.py
│   ├── test_services.py
│   ├── test_use_cases.py
│   ├── test_views.py
│   └── factories.py       # Factory Boy factories
└── migrations/

内置资源

references/ - 全面的 Clean Architecture 和 API 文档

架构模式

• references/project-structure.md 目录布局、文件命名、模块组织、依赖规则

• references/use-cases-pattern.md 输入/输出 DTO、构造函数注入、事务处理、错误处理

• references/services-pattern.md 静态方法与实例方法、验证模式、跨服务通信

• references/models-domain.md 模型中的领域逻辑、TextChoices、UUID 键、索引

• references/views-serializers.md 轻量视图、权限、读取与写入序列化器

• references/testing-clean-arch.md 用例单元测试、模拟服务、集成测试

• references/examples.md 完整 CRUD 示例、复杂工作流、测试示例

API 最佳实践

• references/query-optimization.md N+1 查询、select_related/prefetch_related、索引、批量操作、聚合

• references/api-patterns.md 分页、筛选、搜索、排序、API 版本控制、错误处理、限流

• references/authentication.md JWT 身份验证、权限、API 密钥、基于角色的访问控制、安全标头

• references/production-api.md 设置结构、数据库配置、缓存、日志记录、Sentry、健康检查、Docker

• references/django-admin.md ModelAdmin 配置、内联、定制操作、查询优化、权限、安全

代码质量

• references/coding-standards.md Ruff 配置、导入排序、安全实践、YAGNI、日志记录、文档

核心原则

业务逻辑不使用异常

# BAD
def create_order(self, data):
    if not self.can_create():
        raise ValidationError("Cannot create order")
    return Order.objects.create(**data)

# GOOD
def create_order(self, data) -> CreateOrderOutput:
    if not self.can_create():
        return CreateOrderOutput(success=False, error="Cannot create order")
    order = Order.objects.create(**data)
    return CreateOrderOutput(success=True, order_id=order.id)

验证返回元组

# BAD
def validate(self, order) -> bool:
    return order.status != 'shipped'

# GOOD
def validate(self, order) -> tuple[bool, str]:
    if order.status == 'shipped':
        return False, "Cannot modify shipped orders"
    return True, ""

使用构造函数注入以便测试

# BAD - Hard to test
class CreateOrderUseCase:
    def execute(self, input_dto):
        service = OrderService()  # Hardcoded dependency
        return service.create(input_dto)

# GOOD - Easy to mock
class CreateOrderUseCase:
    def __init__(self, order_service: OrderService):
        self._order_service = order_service

    def execute(self, input_dto):
        return self._order_service.create(input_dto)

补充说明

Python 版本： 必须为 3.10+（支持 X | None 联合类型语法），推荐使用 3.12+ Django 版本： 最低要求为 4.2 LTS，推荐使用 5.0+

命名约定：

• 用例：<action>_<entity>.py（例如，create_order.py）
• 服务：<entity>_service.py（例如，order_service.py）
• 序列化器：<Entity>ReadSerializer、<Entity>CreateSerializer

与其他技能集成：

• 使用 django-celery-expert 处理后台任务
• 使用 refactordjango 将现有代码迁移到此架构