django-drf

关键模式

• 始终按操作区分序列化器：读取 / 创建 / 更新 / 包含
• 始终对复杂筛选使用 filterset_class（而非 filterset_fields）
• 始终在写入序列化器中验证未知字段（继承 BaseWriteSerializer）
• 始终在 get_queryset() 中使用 select_related/prefetch_related，以避免 N+1 查询
• 始终在 get_queryset() 中处理 swagger_fake_view，以支持架构生成
• 始终对 SerializerMethodField 使用 @extend_schema_field 编写 OpenAPI 文档
• 切勿在序列化器中编写业务逻辑，应使用服务/工具函数
• 切勿使用自增主键，应使用 UUIDv4 或 UUIDv7
• 切勿在 URL 中使用尾部斜杠（trailing_slash=False）

注意： swagger_fake_view 是 drf-spectacular 专用于生成 OpenAPI 架构的功能。

---

实施检查清单

实现新端点时，请按以下顺序检查这些模式：

#	模式	参考	要点
1	模型	api/models.py	UUID 主键、inserted_at/updated_at、JSONAPIMeta.resource_name
2	ViewSet	api/base_views.py、api/v1/views.py	继承 BaseRLSViewSet，在 get_queryset() 中防止 N+1 查询
3	序列化器	api/v1/serializers.py	区分读取/创建/更新/包含，继承 BaseWriteSerializer
4	筛选器	api/filters.py	使用 filterset_class，继承基础筛选器类
5	权限	api/base_views.py	required_permissions、set_required_permissions()
6	分页	api/pagination.py	必要时使用自定义分页类
7	URL 路由	api/v1/urls.py	trailing_slash=False，路径采用 kebab-case
8	OpenAPI 架构	api/v1/views.py	使用 drf-spectacular 的 @extend_schema_view
9	测试	api/tests/test_views.py	JSON:API 内容类型、fixture 模式

完整文件路径：请参阅 references/file-locations.md

---

决策树

使用哪种序列化器？

GET list/retrieve → <Model>Serializer
POST create       → <Model>CreateSerializer
PATCH update      → <Model>UpdateSerializer
?include=...      → <Model>IncludeSerializer

使用哪种基础序列化器？

Read-only serializer   → BaseModelSerializerV1
Create with tenant_id  → RLSSerializer + BaseWriteSerializer (auto-injects tenant_id on create)
Update with validation → BaseWriteSerializer (tenant_id already exists on object)
Non-model data         → BaseSerializerV1

使用哪种基础筛选器？

Direct FK to Provider  → BaseProviderFilter
FK via Scan           → BaseScanProviderFilter
No provider relation  → FilterSet

使用哪种基础 ViewSet？

RLS-protected model  → BaseRLSViewSet (most common)
Tenant operations    → BaseTenantViewset
User operations      → BaseUserViewset
No RLS required      → BaseViewSet (rare)

资源名称格式？

Single word model     → plural lowercase           (Provider → providers)
Multi-word model      → plural lowercase kebab     (ProviderGroup → provider-groups)
Through/join model    → parent-child pattern       (UserRoleRelationship → user-roles)
Aggregation/overview  → descriptive kebab plural   (ComplianceOverview → compliance-overviews)

---

序列化器模式

基类层级

# Read serializer (most common)
class ProviderSerializer(RLSSerializer):
    class Meta:
        model = Provider
        fields = ["id", "provider", "uid", "alias", "connected", "inserted_at"]

# Write serializer (validates unknown fields)
class ProviderCreateSerializer(RLSSerializer, BaseWriteSerializer):
    class Meta:
        model = Provider
        fields = ["provider", "uid", "alias"]

# Include serializer (sparse fields for ?include=)
class ProviderIncludeSerializer(RLSSerializer):
    class Meta:
        model = Provider
        fields = ["id", "alias"]  # Minimal fields

带有 OpenAPI 文档的 SerializerMethodField

from drf_spectacular.utils import extend_schema_field

class ProviderSerializer(RLSSerializer):
    connection = serializers.SerializerMethodField(read_only=True)

    @extend_schema_field({
        "type": "object",
        "properties": {
            "connected": {"type": "boolean"},
            "last_checked_at": {"type": "string", "format": "date-time"},
        },
    })
    def get_connection(self, obj):
        return {
            "connected": obj.connected,
            "last_checked_at": obj.connection_last_checked_at,
        }

包含的序列化器（JSON:API）

class ScanSerializer(RLSSerializer):
    included_serializers = {
        "provider": "api.v1.serializers.ProviderIncludeSerializer",
    }

敏感数据脱敏

def to_representation(self, instance):
    data = super().to_representation(instance)
    # Mask by default, expose only on explicit request
    fields_param = self.context.get("request").query_params.get("fields[my-model]", "")
    if "api_key" in fields_param:
        data["api_key"] = instance.api_key_decoded
    else:
        data["api_key"] = "****" if instance.api_key else None
    return data

---

ViewSet 模式

带 N+1 查询防护的 get_queryset()

始终将 swagger_fake_view 检查与 select_related/prefetch_related 结合使用：

def get_queryset(self):
    # REQUIRED: Return empty queryset for OpenAPI schema generation
    if getattr(self, "swagger_fake_view", False):
        return Provider.objects.none()

    # N+1 prevention: eager load relationships
    return Provider.objects.select_related(
        "tenant",
    ).prefetch_related(
        "provider_groups",
        Prefetch("tags", queryset=ProviderTag.objects.filter(tenant_id=self.request.tenant_id)),
    )

为何使用 swagger_fake_view？ drf-spectacular 会检查 ViewSet 以生成 OpenAPI 架构。如果没有此检查，它会执行真实查询，并可能因缺少请求上下文而失败。

按操作指定序列化器

def get_serializer_class(self):
    if self.action == "create":
        return ProviderCreateSerializer
    elif self.action == "partial_update":
        return ProviderUpdateSerializer
    elif self.action in ["connection", "destroy"]:
        return TaskSerializer
    return ProviderSerializer

按操作动态设置权限

class ProviderViewSet(BaseRLSViewSet):
    required_permissions = [Permissions.MANAGE_PROVIDERS]

    def set_required_permissions(self):
        if self.action in ["list", "retrieve"]:
            self.required_permissions = []  # Read-only = no permission
        else:
            self.required_permissions = [Permissions.MANAGE_PROVIDERS]

缓存装饰器

from django.utils.decorators import method_decorator
from django.views.decorators.cache import cache_control

CACHE_DECORATOR = cache_control(
    max_age=django_settings.CACHE_MAX_AGE,
    stale_while_revalidate=django_settings.CACHE_STALE_WHILE_REVALIDATE,
)

@method_decorator(CACHE_DECORATOR, name="list")
@method_decorator(CACHE_DECORATOR, name="retrieve")
class ProviderViewSet(BaseRLSViewSet):
    pass

自定义操作

# Detail action (operates on single object)
@action(detail=True, methods=["post"], url_name="connection")
def connection(self, request, pk=None):
    instance = self.get_object()
    # Process instance...

# List action (operates on collection)
@action(detail=False, methods=["get"], url_name="metadata")
def metadata(self, request):
    queryset = self.filter_queryset(self.get_queryset())
    # Aggregate over queryset...

---

筛选器模式

基础筛选器类

class BaseProviderFilter(FilterSet):
    """For models with direct FK to Provider"""
    provider_id = UUIDFilter(field_name="provider__id", lookup_expr="exact")
    provider_id__in = UUIDInFilter(field_name="provider__id", lookup_expr="in")
    provider_type = ChoiceFilter(field_name="provider__provider", choices=Provider.ProviderChoices.choices)

class BaseScanProviderFilter(FilterSet):
    """For models with FK to Scan (Scan has FK to Provider)"""
    provider_id = UUIDFilter(field_name="scan__provider__id", lookup_expr="exact")

自定义多值筛选器

class UUIDInFilter(BaseInFilter, UUIDFilter):
    pass

class CharInFilter(BaseInFilter, CharFilter):
    pass

class ChoiceInFilter(BaseInFilter, ChoiceFilter):
    pass

ArrayField 筛选

# Single value contains
region = CharFilter(method="filter_region")

def filter_region(self, queryset, name, value):
    return queryset.filter(resource_regions__contains=[value])

# Multi-value overlap
region__in = CharInFilter(field_name="resource_regions", lookup_expr="overlap")

日期范围验证

def filter_queryset(self, queryset):
    # Require date filter for performance
    if not (date_filters_provided):
        raise ValidationError([{
            "detail": "At least one date filter is required",
            "status": 400,
            "source": {"pointer": "/data/attributes/inserted_at"},
            "code": "required",
        }])

    # Validate max range
    if date_range > settings.FINDINGS_MAX_DAYS_IN_RANGE:
        raise ValidationError(...)

    return super().filter_queryset(queryset)

动态选择 FilterSet

def get_filterset_class(self):
    if self.action in ["latest", "metadata_latest"]:
        return LatestFindingFilter
    return FindingFilter

枚举字段覆盖

class Meta:
    model = Finding
    filter_overrides = {
        FindingDeltaEnumField: {"filter_class": CharFilter},
        StatusEnumField: {"filter_class": CharFilter},
        SeverityEnumField: {"filter_class": CharFilter},
    }

---

性能模式

PaginateByPkMixin

适用于包含开销较大的联接的大型查询集：

class PaginateByPkMixin:
    def paginate_by_pk(self, request, base_queryset, manager,
                       select_related=None, prefetch_related=None):
        # 1. Get PKs only (cheap)
        pk_list = base_queryset.values_list("id", flat=True)
        page = self.paginate_queryset(pk_list)

        # 2. Fetch full objects for just the page
        queryset = manager.filter(id__in=page)
        if select_related:
            queryset = queryset.select_related(*select_related)
        if prefetch_related:
            queryset = queryset.prefetch_related(*prefetch_related)

        # 3. Re-sort to preserve DB ordering
        queryset = sorted(queryset, key=lambda obj: page.index(obj.id))
        return self.get_paginated_response(self.get_serializer(queryset, many=True).data)

在序列化器中预取

def get_tags(self, obj):
    # Use prefetched tags if available
    if hasattr(obj, "prefetched_tags"):
        return {tag.key: tag.value for tag in obj.prefetched_tags}
    # Fallback (causes N+1 if not prefetched)
    return obj.get_tags(self.context.get("tenant_id"))

---

命名约定

实体	模式	示例
序列化器（读取）	<Model>Serializer	ProviderSerializer
序列化器（创建）	<Model>CreateSerializer	ProviderCreateSerializer
序列化器（更新）	<Model>UpdateSerializer	ProviderUpdateSerializer
序列化器（包含）	<Model>IncludeSerializer	ProviderIncludeSerializer
筛选器	<Model>Filter	ProviderFilter
ViewSet	<Model>ViewSet	ProviderViewSet

---

OpenAPI 文档

from drf_spectacular.utils import extend_schema, extend_schema_view

@extend_schema_view(
    list=extend_schema(tags=["Provider"], summary="List all providers"),
    retrieve=extend_schema(tags=["Provider"], summary="Retrieve provider"),
    create=extend_schema(tags=["Provider"], summary="Create provider"),
)
@extend_schema(tags=["Provider"])
class ProviderViewSet(BaseRLSViewSet):
    pass

---

API 安全模式

完整示例：参见 assets/security_patterns.py

模式	关键要点
输入验证	使用 validate_<field>() 进行清理，使用 validate() 进行跨字段验证
防止批量赋值	始终使用显式的 fields 列表，绝不要使用 __all__ 或 exclude
对象级权限	实现 has_object_permission() 以检查所有权
速率限制	配置 DEFAULT_THROTTLE_RATES，对敏感端点使用视图级限流
防止信息泄露	使用通用错误消息；未经授权时返回 404 而不是 403（防止枚举）
SQL 注入	始终使用 ORM 参数化，绝不要在原始 SQL 中使用字符串插值

快速参考

# Input validation in serializer
def validate_uid(self, value):
    value = value.strip().lower()
    if not re.match(r'^[a-z0-9-]+$', value):
        raise serializers.ValidationError("Invalid format")
    return value

# Explicit fields (prevent mass assignment)
class Meta:
    fields = ["name", "email"]  # GOOD: whitelist
    read_only_fields = ["id", "inserted_at"]  # System fields

# Object permission
class IsOwnerOrReadOnly(BasePermission):
    def has_object_permission(self, request, view, obj):
        if request.method in SAFE_METHODS:
            return True
        return obj.owner == request.user

# Throttling for sensitive endpoints
class BurstRateThrottle(UserRateThrottle):
    rate = "10/minute"

# Safe error messages (prevent enumeration)
def get_object(self):
    try:
        return super().get_object()
    except Http404:
        raise NotFound("Resource not found")  # Generic, no internal IDs

---

命令

# Development
cd api && uv run python src/backend/manage.py runserver
cd api && uv run python src/backend/manage.py shell

# Database
cd api && uv run python src/backend/manage.py makemigrations
cd api && uv run python src/backend/manage.py migrate

# Testing
cd api && uv run pytest -x --tb=short
cd api && uv run make lint

---

资源

本地参考

• 文件位置：参见 references/file-locations.md
• JSON:API 约定：参见 references/json-api-conventions.md
• 安全模式：参见 assets/security_patterns.py

Context7 MCP（推荐）

**前提条件：**安装 Context7 MCP 服务，以便查询最新文档。

在实现或调试时，通过 mcp_context7_query-docs 查询以下库：

库	Context7 ID	用途
Django	/websites/djangoproject_en_5_2	模型、ORM、迁移
DRF	/websites/django-rest-framework	ViewSet、序列化器、权限
drf-spectacular	/tfranzel/drf-spectacular	OpenAPI 架构、@extend_schema

示例查询：

mcp_context7_query-docs(libraryId="/websites/django-rest-framework", query="ViewSet get_queryset best practices")
mcp_context7_query-docs(libraryId="/tfranzel/drf-spectacular", query="extend_schema examples for custom actions")
mcp_context7_query-docs(libraryId="/websites/djangoproject_en_5_2", query="model constraints and indexes")

**注意：**如果需要查找正确的库 ID，请先使用 mcp_context7_resolve-library-id。

外部文档

• DRF 文档：https://www.django-rest-framework.org/
• DRF JSON:API：https://django-rest-framework-json-api.readthedocs.io/
• drf-spectacular：https://drf-spectacular.readthedocs.io/
• django-filter：https://django-filter.readthedocs.io/