高效代码生成

下面给出一个可直接运行的、可扩展的 Django + DRF 任务管理 REST 后端模板，覆盖需求中的功能、架构与工程实践。包含项目目录结构、关键代码文件、配置、Docker、测试、运行步骤与扩展指导。可作为生产化项目的起点。

目录结构

• task_manager/
  • manage.py
  • requirements.txt
  • Dockerfile
  • docker-compose.yml
  • .env.example
  • README.md
  • config/
    • init.py
    • asgi.py
    • wsgi.py
    • urls.py
    • settings/
      • init.py
      • base.py
      • dev.py
      • staging.py
      • prod.py
      • logging.py
  • apps/
    • common/
      • init.py
      • middleware.py
      • pagination.py
      • permissions.py
      • utils.py
      • cache.py
    • users/
      • init.py
      • models.py
      • serializers.py
      • repositories.py
      • services.py
      • views.py
      • permissions.py
      • urls.py
    • projects/
      • init.py
      • models.py
      • serializers.py
      • repositories.py
      • services.py
      • views.py
      • urls.py
    • tasks/
      • init.py
      • choices.py
      • models.py
      • validators.py
      • dtos.py
      • serializers.py
      • repositories.py
      • factories.py
      • strategies.py
      • observers.py
      • services.py
      • views.py
      • urls.py
      • management/
        • init.py
        • commands/
          • init.py
          • seed.py
          • remind_due.py
  • tests/
    • conftest.py
    • test_models.py
    • test_services.py
    • test_api.py

requirements.txt

Django>=4.2
djangorestframework>=3.14
djangorestframework-simplejwt>=5.3
drf-spectacular>=0.27
django-filter>=23.5
python-dotenv>=1.0
dj-database-url>=2.2
psycopg2-binary>=2.9
gunicorn>=21
uvicorn>=0.30
whitenoise>=6.7
django-cors-headers>=4.4
requests>=2.32
django-redis>=5.4
python-json-logger>=2.0
pytest>=8.2
pytest-django>=4.8
pytest-cov>=5.0
Faker>=26.0

config/settings/base.py

import os
from pathlib import Path
from datetime import timedelta
from dotenv import load_dotenv
import dj_database_url

load_dotenv()

BASE_DIR = Path(__file__).resolve().parent.parent.parent
SECRET_KEY = os.getenv("SECRET_KEY", "dev-insecure-key")
DEBUG = os.getenv("DEBUG", "True") == "True"
ALLOWED_HOSTS = os.getenv("ALLOWED_HOSTS", "*").split(",")

INSTALLED_APPS = [
    "django.contrib.admin",
    "django.contrib.auth",
    "django.contrib.contenttypes",
    "django.contrib.sessions",
    "django.contrib.messages",
    "django.contrib.staticfiles",
    "rest_framework",
    "rest_framework.authtoken",
    "drf_spectacular",
    "django_filters",
    "corsheaders",
    "apps.users",
    "apps.projects",
    "apps.tasks",
]

MIDDLEWARE = [
    "corsheaders.middleware.CorsMiddleware",
    "django.middleware.security.SecurityMiddleware",
    "whitenoise.middleware.WhiteNoiseMiddleware",
    "django.contrib.sessions.middleware.SessionMiddleware",
    "django.middleware.common.CommonMiddleware",
    "django.middleware.csrf.CsrfViewMiddleware",
    "django.contrib.auth.middleware.AuthenticationMiddleware",
    "django.contrib.messages.middleware.MessageMiddleware",
    "django.middleware.clickjacking.XFrameOptionsMiddleware",
    "apps.common.middleware.RequestIDMiddleware",
    "apps.common.middleware.StructuredExceptionMiddleware",
]

ROOT_URLCONF = "config.urls"
WSGI_APPLICATION = "config.wsgi.application"
ASGI_APPLICATION = "config.asgi.application"

DATABASES = {
    "default": dj_database_url.parse(
        os.getenv("DATABASE_URL", f"sqlite:///{BASE_DIR / 'db.sqlite3'}"),
        conn_max_age=600,
    )
}

AUTH_USER_MODEL = "users.User"

REST_FRAMEWORK = {
    "DEFAULT_AUTHENTICATION_CLASSES": (
        "rest_framework_simplejwt.authentication.JWTAuthentication",
    ),
    "DEFAULT_PERMISSION_CLASSES": (
        "rest_framework.permissions.IsAuthenticated",
    ),
    "DEFAULT_SCHEMA_CLASS": "drf_spectacular.openapi.AutoSchema",
    "DEFAULT_PAGINATION_CLASS": "apps.common.pagination.DefaultPageNumberPagination",
    "PAGE_SIZE": int(os.getenv("PAGE_SIZE", 20)),
    "DEFAULT_FILTER_BACKENDS": [
        "django_filters.rest_framework.DjangoFilterBackend",
        "rest_framework.filters.OrderingFilter",
        "rest_framework.filters.SearchFilter",
    ],
    "DEFAULT_THROTTLE_RATES": {
        "anon": os.getenv("THROTTLE_ANON", "100/min"),
        "user": os.getenv("THROTTLE_USER", "1000/min"),
    },
}

SIMPLE_JWT = {
    "ACCESS_TOKEN_LIFETIME": timedelta(minutes=int(os.getenv("JWT_ACCESS_MIN", "60"))),
    "REFRESH_TOKEN_LIFETIME": timedelta(days=int(os.getenv("JWT_REFRESH_DAYS", "7"))),
    "AUTH_HEADER_TYPES": ("Bearer",),
}

SPECTACULAR_SETTINGS = {
    "TITLE": "Task Manager API",
    "DESCRIPTION": "Django + DRF 任务管理后端",
    "VERSION": "1.0.0",
    "SERVE_PERMISSIONS": ["rest_framework.permissions.AllowAny"],
}

LANGUAGE_CODE = "zh-hans"
TIME_ZONE = os.getenv("TIME_ZONE", "Asia/Shanghai")
USE_I18N = True
USE_TZ = True

STATIC_URL = "static/"
STATIC_ROOT = BASE_DIR / "static"
MEDIA_URL = "/media/"
MEDIA_ROOT = BASE_DIR / "media"

CORS_ALLOW_ALL_ORIGINS = True

CACHES = {
    "default": {
        "BACKEND": "django_redis.cache.RedisCache"
        if os.getenv("REDIS_URL")
        else "django.core.cache.backends.locmem.LocMemCache",
        "LOCATION": os.getenv("REDIS_URL", ""),
        "OPTIONS": {"CLIENT_CLASS": "django_redis.client.DefaultClient"} if os.getenv("REDIS_URL") else {},
    }
}

LOGGING = {
    "version": 1,
    "disable_existing_loggers": False,
    "formatters": {
        "json": {
            "()": "python_json_logger.jsonlogger.JsonFormatter",
            "fmt": "%(asctime)s %(levelname)s %(name)s %(message)s %(request_id)s",
        },
    },
    "handlers": {
        "console": {"class": "logging.StreamHandler", "formatter": "json"},
    },
    "root": {"handlers": ["console"], "level": "INFO"},
}

FILE_UPLOAD_HANDLERS = ["django.core.files.uploadhandler.TemporaryFileUploadHandler"]

# Feature flags
FEATURE_RATE_LIMIT = os.getenv("FEATURE_RATE_LIMIT", "True") == "True"

config/settings/dev.py

from .base import *
DEBUG = True

config/settings/staging.py

from .base import *
DEBUG = False

config/settings/prod.py

from .base import *
DEBUG = False
SECURE_PROXY_SSL_HEADER = ("HTTP_X_FORWARDED_PROTO", "https")
CSRF_TRUSTED_ORIGINS = os.getenv("CSRF_TRUSTED_ORIGINS", "").split(",") if os.getenv("CSRF_TRUSTED_ORIGINS") else []

config/urls.py

from django.contrib import admin
from django.urls import path, include
from drf_spectacular.views import SpectacularAPIView, SpectacularSwaggerView

urlpatterns = [
    path("admin/", admin.site.urls),
    path("api/schema/", SpectacularAPIView.as_view(), name="schema"),
    path("api/docs/", SpectacularSwaggerView.as_view(url_name="schema")),
    path("auth/", include("apps.users.urls")),
    path("projects/", include("apps.projects.urls")),
    path("tasks/", include("apps.tasks.urls")),
]

apps/common/pagination.py

from rest_framework.pagination import PageNumberPagination

class DefaultPageNumberPagination(PageNumberPagination):
    page_size_query_param = "page_size"
    max_page_size = 200

apps/common/middleware.py

import uuid, logging, json
from django.utils.deprecation import MiddlewareMixin
from django.http import JsonResponse

logger = logging.getLogger(__name__)

class RequestIDMiddleware(MiddlewareMixin):
    def process_request(self, request):
        request.request_id = str(uuid.uuid4())

class StructuredExceptionMiddleware(MiddlewareMixin):
    def process_exception(self, request, exception):
        logger.exception("Unhandled exception", extra={"request_id": getattr(request, "request_id", None)})
        return JsonResponse(
            {"detail": "Internal Server Error", "request_id": getattr(request, "request_id", None)},
            status=500,
        )

apps/common/permissions.py

from rest_framework.permissions import BasePermission, SAFE_METHODS

class IsAdmin(BasePermission):
    def has_permission(self, request, view):
        return request.user.is_authenticated and request.user.role == "admin"

class IsOwnerOrAdmin(BasePermission):
    def has_object_permission(self, request, view, obj):
        if not request.user.is_authenticated:
            return False
        return request.user.role == "admin" or getattr(obj, "owner_id", None) == request.user.id

apps/users/models.py

from django.contrib.auth.models import AbstractUser
from django.db import models

class User(AbstractUser):
    ROLE_CHOICES = (("admin", "Admin"), ("member", "Member"))
    role = models.CharField(max_length=10, choices=ROLE_CHOICES, default="member")

    def __str__(self):
        return f"{self.username}({self.role})"

apps/users/serializers.py

from rest_framework import serializers
from django.contrib.auth import get_user_model
from django.contrib.auth.password_validation import validate_password

User = get_user_model()

class UserDTO(serializers.Serializer):
    id = serializers.IntegerField(read_only=True)
    username = serializers.CharField()
    email = serializers.EmailField()
    role = serializers.CharField()

class RegisterSerializer(serializers.ModelSerializer):
    password = serializers.CharField(write_only=True)
    class Meta:
        model = User
        fields = ("id", "username", "email", "password", "role")

    def validate_password(self, value):
        validate_password(value)
        return value

    def create(self, validated_data):
        password = validated_data.pop("password")
        user = User(**validated_data)
        user.set_password(password)
        user.save()
        return user

class LoginSerializer(serializers.Serializer):
    username = serializers.CharField()
    password = serializers.CharField(write_only=True)

apps/users/repositories.py

from django.contrib.auth import get_user_model

User = get_user_model()

class UserRepository:
    @staticmethod
    def get_by_username(username: str):
        try:
            return User.objects.get(username=username)
        except User.DoesNotExist:
            return None

apps/users/services.py

from django.contrib.auth import authenticate
from rest_framework.exceptions import AuthenticationFailed

class AuthService:
    @staticmethod
    def authenticate_user(username: str, password: str):
        user = authenticate(username=username, password=password)
        if not user:
            raise AuthenticationFailed("Invalid credentials")
        return user

apps/users/views.py

from rest_framework import generics, status
from rest_framework.response import Response
from rest_framework_simplejwt.tokens import RefreshToken
from .serializers import RegisterSerializer, LoginSerializer, UserDTO
from .services import AuthService

class RegisterView(generics.CreateAPIView):
    serializer_class = RegisterSerializer
    permission_classes = []  # AllowAny

class LoginView(generics.GenericAPIView):
    serializer_class = LoginSerializer
    permission_classes = []  # AllowAny

    def post(self, request, *args, **kwargs):
        ser = self.get_serializer(data=request.data)
        ser.is_valid(raise_exception=True)
        user = AuthService.authenticate_user(ser.validated_data["username"], ser.validated_data["password"])
        refresh = RefreshToken.for_user(user)
        return Response(
            {"access": str(refresh.access_token), "refresh": str(refresh), "user": UserDTO(user).data},
            status=status.HTTP_200_OK
        )

apps/users/urls.py

from django.urls import path
from .views import RegisterView, LoginView

urlpatterns = [
    path("register", RegisterView.as_view(), name="register"),
    path("login", LoginView.as_view(), name="login"),
]

apps/projects/models.py

from django.db import models
from django.conf import settings

class Project(models.Model):
    name = models.CharField(max_length=255)
    description = models.TextField(blank=True)
    owner = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="owned_projects")
    created_at = models.DateTimeField(auto_now_add=True)

    def __str__(self):
        return self.name

apps/projects/serializers.py

from rest_framework import serializers
from .models import Project

class ProjectSerializer(serializers.ModelSerializer):
    owner_username = serializers.ReadOnlyField(source="owner.username")
    class Meta:
        model = Project
        fields = ("id", "name", "description", "owner", "owner_username", "created_at")
        read_only_fields = ("owner", "created_at")

apps/projects/repositories.py

from .models import Project
from django.core.cache import cache

class ProjectRepository:
    @staticmethod
    def list_owned_or_admin(user, search=None):
        qs = Project.objects.all() if user.role == "admin" else Project.objects.filter(owner=user)
        if search:
            qs = qs.filter(name__icontains=search)
        return qs.order_by("-created_at")

    @staticmethod
    def get(pk: int):
        return Project.objects.get(pk=pk)

apps/projects/services.py

from .repositories import ProjectRepository

class ProjectService:
    @staticmethod
    def list_projects(user, search=None):
        return ProjectRepository.list_owned_or_admin(user, search=search)

apps/projects/views.py

from rest_framework import viewsets, permissions
from .serializers import ProjectSerializer
from .models import Project
from apps.common.permissions import IsOwnerOrAdmin

class ProjectViewSet(viewsets.ModelViewSet):
    serializer_class = ProjectSerializer
    queryset = Project.objects.all()

    def get_permissions(self):
        if self.action in ["list", "retrieve", "create"]:
            return [permissions.IsAuthenticated()]
        return [IsOwnerOrAdmin()]

    def get_queryset(self):
        search = self.request.query_params.get("search")
        from .services import ProjectService
        return ProjectService.list_projects(self.request.user, search=search)

    def perform_create(self, serializer):
        serializer.save(owner=self.request.user)

apps/projects/urls.py

from rest_framework.routers import DefaultRouter
from .views import ProjectViewSet

router = DefaultRouter()
router.register(r"", ProjectViewSet, basename="project")

urlpatterns = router.urls

apps/tasks/choices.py

PRIORITY_CHOICES = (("低", "低"), ("中", "中"), ("高", "高"))
STATUS_CHOICES = (("TODO", "TODO"), ("DOING", "DOING"), ("DONE", "DONE"))

apps/tasks/models.py

from django.db import models
from django.conf import settings
from .choices import PRIORITY_CHOICES, STATUS_CHOICES

class Label(models.Model):
    name = models.CharField(max_length=50)
    parent = models.ForeignKey("self", null=True, blank=True, on_delete=models.SET_NULL, related_name="children")
    def __str__(self): return self.name

class Task(models.Model):
    title = models.CharField(max_length=255)
    description = models.TextField(blank=True)
    priority = models.CharField(max_length=2, choices=PRIORITY_CHOICES, default="中")
    status = models.CharField(max_length=5, choices=STATUS_CHOICES, default="TODO")
    due_date = models.DateTimeField(null=True, blank=True)
    assignee = models.ForeignKey(settings.AUTH_USER_MODEL, null=True, blank=True, on_delete=models.SET_NULL, related_name="assigned_tasks")
    project = models.ForeignKey("projects.Project", on_delete=models.CASCADE, related_name="tasks")
    labels = models.ManyToManyField(Label, blank=True, related_name="tasks")
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

class Attachment(models.Model):
    task = models.ForeignKey(Task, on_delete=models.CASCADE, related_name="attachments")
    file = models.FileField(upload_to="attachments/")
    uploaded_at = models.DateTimeField(auto_now_add=True)

class Comment(models.Model):
    task = models.ForeignKey(Task, on_delete=models.CASCADE, related_name="comments")
    author = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
    content = models.TextField()
    created_at = models.DateTimeField(auto_now_add=True)

class Webhook(models.Model):
    EVENT_CHOICES = (("task.status_changed", "task.status_changed"), ("task.created", "task.created"))
    url = models.URLField()
    event = models.CharField(max_length=50, choices=EVENT_CHOICES)
    active = models.BooleanField(default=True)
    secret = models.CharField(max_length=128, blank=True, default="")
    created_at = models.DateTimeField(auto_now_add=True)

apps/tasks/validators.py

from rest_framework import serializers

class CSVImportValidator(serializers.Serializer):
    file = serializers.FileField()
    project_id = serializers.IntegerField(required=False)
    default_assignee = serializers.IntegerField(required=False)

apps/tasks/dtos.py

from dataclasses import dataclass
from typing import Optional, List

@dataclass
class TaskDTO:
    title: str
    description: str
    priority: str
    status: str
    due_date: Optional[str]
    assignee_id: Optional[int]
    project_id: int
    labels: Optional[list]

apps/tasks/serializers.py

from rest_framework import serializers
from .models import Task, Comment, Attachment, Label
from .choices import PRIORITY_CHOICES, STATUS_CHOICES

class LabelSerializer(serializers.ModelSerializer):
    class Meta:
        model = Label
        fields = ("id", "name", "parent")

class AttachmentSerializer(serializers.ModelSerializer):
    class Meta:
        model = Attachment
        fields = ("id", "file", "uploaded_at")

class CommentSerializer(serializers.ModelSerializer):
    author_username = serializers.ReadOnlyField(source="author.username")
    class Meta:
        model = Comment
        fields = ("id", "author", "author_username", "content", "created_at")
        read_only_fields = ("author", "created_at")

class TaskSerializer(serializers.ModelSerializer):
    assignee_username = serializers.ReadOnlyField(source="assignee.username")
    labels = LabelSerializer(many=True, required=False)
    attachments = AttachmentSerializer(many=True, read_only=True)

    class Meta:
        model = Task
        fields = (
            "id", "title", "description", "priority", "status", "due_date",
            "assignee", "assignee_username", "labels", "project", "attachments",
            "created_at", "updated_at"
        )
        read_only_fields = ("created_at", "updated_at")

    def create(self, validated_data):
        labels_data = validated_data.pop("labels", [])
        task = Task.objects.create(**validated_data)
        self._upsert_labels(task, labels_data)
        return task

    def update(self, instance, validated_data):
        labels_data = validated_data.pop("labels", None)
        for k, v in validated_data.items():
            setattr(instance, k, v)
        instance.save()
        if labels_data is not None:
            self._upsert_labels(instance, labels_data, replace=True)
        return instance

    def _upsert_labels(self, task, labels_data, replace=False):
        label_objs = []
        for ld in labels_data:
            obj, _ = Label.objects.get_or_create(name=ld.get("name"), defaults={"parent": ld.get("parent")})
            label_objs.append(obj)
        if replace:
            task.labels.set(label_objs)
        else:
            task.labels.add(*label_objs)

apps/tasks/repositories.py

from .models import Task, Comment, Attachment, Webhook
from django.db.models import Q
from django.core.cache import cache

class TaskRepository:
    @staticmethod
    def list_filtered(user, params):
        qs = Task.objects.select_related("assignee", "project").prefetch_related("labels")
        # Permissions: admin sees all; member sees own assigned + own projects
        if user.role != "admin":
            qs = qs.filter(Q(assignee=user) | Q(project__owner=user))
        status_ = params.get("status")
        priority = params.get("priority")
        project = params.get("project")
        search = params.get("search")
        if status_:
            qs = qs.filter(status=status_)
        if priority:
            qs = qs.filter(priority=priority)
        if project:
            qs = qs.filter(project_id=project)
        if search:
            qs = qs.filter(Q(title__icontains=search) | Q(description__icontains=search))
        ordering = params.get("ordering") or "-created_at"
        qs = qs.order_by(ordering)
        return qs

    @staticmethod
    def get(pk: int):
        return Task.objects.get(pk=pk)

class CommentRepository:
    @staticmethod
    def create(task, author, content):
        return task.comments.create(author=author, content=content)

class AttachmentRepository:
    @staticmethod
    def add(task, file):
        return task.attachments.create(file=file)

class WebhookRepository:
    @staticmethod
    def active_for_event(event: str):
        return Webhook.objects.filter(active=True, event=event)

apps/tasks/factories.py

from .models import Task, Label

class TaskFactory:
    @staticmethod
    def create_default(project, assignee=None):
        task = Task.objects.create(
            title="新任务",
            description="默认描述",
            priority="中",
            status="TODO",
            project=project,
            assignee=assignee,
        )
        default_labels = ["待规划", "快速处理"]
        for name in default_labels:
            lbl, _ = Label.objects.get_or_create(name=name)
            task.labels.add(lbl)
        return task

apps/tasks/strategies.py

from typing import Protocol
from django.db.models import QuerySet

class SortStrategy(Protocol):
    def apply(self, qs: QuerySet) -> QuerySet: ...

class SortByDueDateAsc:
    def apply(self, qs): return qs.order_by("due_date", "id")

class SortByPriorityDesc:
    # 假定优先级中文：高>中>低，自定义排序
    PRIORITY_RANK = {"高": 3, "中": 2, "低": 1}
    def apply(self, qs):
        from django.db.models import IntegerField, Case, When, Value
        priority_order = Case(
            When(priority="高", then=Value(3)),
            When(priority="中", then=Value(2)),
            When(priority="低", then=Value(1)),
            default=Value(0), output_field=IntegerField()
        )
        return qs.order_by(-priority_order, "-created_at")

STRATEGIES = {
    "due_date_asc": SortByDueDateAsc(),
    "priority_desc": SortByPriorityDesc(),
}

apps/tasks/observers.py

import hmac, hashlib, json, logging, requests
from .repositories import WebhookRepository

logger = logging.getLogger(__name__)

class Observer:
    def notify(self, event: str, payload: dict): ...

class WebhookNotifier(Observer):
    def notify(self, event: str, payload: dict):
        hooks = WebhookRepository.active_for_event(event)
        for hook in hooks:
            headers = {"Content-Type": "application/json"}
            body = json.dumps(payload).encode("utf-8")
            if hook.secret:
                sig = hmac.new(hook.secret.encode(), body, hashlib.sha256).hexdigest()
                headers["X-Signature"] = sig
            try:
                requests.post(hook.url, data=body, headers=headers, timeout=5)
            except Exception:
                logger.exception("Webhook delivery failed")

class LoggerNotifier(Observer):
    def notify(self, event: str, payload: dict):
        logger.info("Event: %s payload=%s", event, payload)

apps/tasks/services.py

from typing import List
from django.utils import timezone
from rest_framework.exceptions import ValidationError, PermissionDenied
from .repositories import TaskRepository, CommentRepository, AttachmentRepository
from .observers import WebhookNotifier, LoggerNotifier
from .strategies import STRATEGIES

ALLOWED_TRANSITIONS = {
    "TODO": {"DOING"},
    "DOING": {"DONE", "TODO"},
    "DONE": set(),
}

class TaskService:
    observers = [LoggerNotifier(), WebhookNotifier()]

    @staticmethod
    def list_tasks(user, params):
        qs = TaskRepository.list_filtered(user, params)
        strategy_key = params.get("sort_strategy")
        if strategy_key and strategy_key in STRATEGIES:
            qs = STRATEGIES[strategy_key].apply(qs)
        return qs

    @classmethod
    def change_status(cls, user, task, new_status: str):
        if user.role != "admin" and user != task.assignee and user != task.project.owner:
            raise PermissionDenied("No permission to change status")
        if new_status not in ALLOWED_TRANSITIONS[task.status]:
            raise ValidationError(f"Invalid status transition {task.status} -> {new_status}")
        old = task.status
        task.status = new_status
        task.save(update_fields=["status", "updated_at"])
        payload = {"task_id": task.id, "old": old, "new": new_status, "at": timezone.now().isoformat()}
        for obs in cls.observers:
            obs.notify("task.status_changed", payload)
        return task

    @staticmethod
    def add_comment(task, author, content: str):
        return CommentRepository.create(task, author, content)

    @staticmethod
    def add_attachment(task, file):
        return AttachmentRepository.add(task, file)

    @staticmethod
    def bulk_import_from_csv(file, project=None, default_assignee=None):
        import csv, io
        decoded = file.read().decode("utf-8")
        f = io.StringIO(decoded)
        reader = csv.DictReader(f)
        from .models import Task
        created = []
        for row in reader:
            t = Task.objects.create(
                title=row.get("title", "Untitled"),
                description=row.get("description", ""),
                priority=row.get("priority", "中"),
                status=row.get("status", "TODO"),
                due_date=row.get("due_date") or None,
                project=project,
                assignee=default_assignee,
            )
            created.append(t.id)
        return created

apps/tasks/views.py

from rest_framework import viewsets, mixins, status, decorators
from rest_framework.response import Response
from rest_framework.parsers import MultiPartParser, FormParser
from django.shortcuts import get_object_or_404
from apps.common.permissions import IsAdmin
from .serializers import TaskSerializer, CommentSerializer, AttachmentSerializer, LabelSerializer
from .models import Task, Comment, Attachment, Label, Webhook
from .validators import CSVImportValidator
from .services import TaskService

class TemplateCRUDViewSet(viewsets.ModelViewSet):
    # Template Method: hooks for pre/post
    def perform_create(self, serializer):
        instance = serializer.save()
        self.after_create(instance)

    def perform_update(self, serializer):
        instance = serializer.save()
        self.after_update(instance)

    def perform_destroy(self, instance):
        super().perform_destroy(instance)
        self.after_destroy(instance)

    def after_create(self, instance): pass
    def after_update(self, instance): pass
    def after_destroy(self, instance): pass

class TaskViewSet(TemplateCRUDViewSet):
    serializer_class = TaskSerializer
    queryset = Task.objects.all()

    def get_queryset(self):
        return TaskService.list_tasks(self.request.user, self.request.query_params)

    @decorators.action(detail=True, methods=["post"])
    def change_status(self, request, pk=None):
        task = self.get_object()
        new_status = request.data.get("status")
        TaskService.change_status(request.user, task, new_status)
        return Response({"status": "ok"})

    @decorators.action(detail=True, methods=["post"], url_path="comments")
    def add_comment(self, request, pk=None):
        task = self.get_object()
        ser = CommentSerializer(data=request.data)
        ser.is_valid(raise_exception=True)
        c = TaskService.add_comment(task, request.user, ser.validated_data["content"])
        return Response(CommentSerializer(c).data, status=status.HTTP_201_CREATED)

    @decorators.action(detail=True, methods=["post"], url_path="attachments", parser_classes=[MultiPartParser, FormParser])
    def add_attachment(self, request, pk=None):
        task = self.get_object()
        file = request.data.get("file")
        att = TaskService.add_attachment(task, file)
        return Response(AttachmentSerializer(att).data, status=status.HTTP_201_CREATED)

    @decorators.action(detail=False, methods=["post"], url_path="import-csv", parser_classes=[MultiPartParser, FormParser])
    def import_csv(self, request):
        val = CSVImportValidator(data=request.data)
        val.is_valid(raise_exception=True)
        from apps.projects.models import Project
        project = get_object_or_404(Project, pk=val.validated_data.get("project_id")) if val.validated_data.get("project_id") else None
        from django.contrib.auth import get_user_model
        User = get_user_model()
        assignee = get_object_or_404(User, pk=val.validated_data.get("default_assignee")) if val.validated_data.get("default_assignee") else None
        ids = TaskService.bulk_import_from_csv(val.validated_data["file"], project, assignee)
        return Response({"created_ids": ids}, status=status.HTTP_201_CREATED)

class LabelViewSet(viewsets.ModelViewSet):
    serializer_class = LabelSerializer
    queryset = Label.objects.all()
    permission_classes = [IsAdmin]

class WebhookViewSet(viewsets.ModelViewSet):
    from rest_framework import permissions
    class Serializer(serializers.ModelSerializer):
        class Meta:
            model = Webhook
            fields = ("id", "url", "event", "active", "secret", "created_at")
            read_only_fields = ("created_at",)
    serializer_class = Serializer
    queryset = Webhook.objects.all()
    permission_classes = [IsAdmin]

apps/tasks/urls.py

from rest_framework.routers import DefaultRouter
from .views import TaskViewSet, LabelViewSet, WebhookViewSet

router = DefaultRouter()
router.register(r"", TaskViewSet, basename="task")
router.register(r"labels", LabelViewSet, basename="label")
router.register(r"webhooks", WebhookViewSet, basename="webhook")

urlpatterns = router.urls

apps/tasks/management/commands/seed.py

from django.core.management.base import BaseCommand
from django.contrib.auth import get_user_model
from faker import Faker
from apps.projects.models import Project
from apps.tasks.factories import TaskFactory

class Command(BaseCommand):
    help = "Seed database with sample data"

    def handle(self, *args, **options):
        fake = Faker("zh_CN")
        User = get_user_model()
        admin, _ = User.objects.get_or_create(username="admin", defaults={"email": "admin@example.com", "role":"admin"})
        if not admin.pk or not admin.has_usable_password():
            admin.set_password("Admin123!")
            admin.save()
        for i in range(3):
            u, _ = User.objects.get_or_create(username=f"user{i}", defaults={"email": f"user{i}@ex.com", "role":"member"})
            if not u.has_usable_password():
                u.set_password("User123!")
                u.save()
        for i in range(3):
            owner = User.objects.order_by("?").first()
            p = Project.objects.create(name=fake.catch_phrase(), description=fake.text(), owner=owner)
            TaskFactory.create_default(p, assignee=owner)
        self.stdout.write(self.style.SUCCESS("Seeding completed"))

apps/tasks/management/commands/remind_due.py

from django.core.management.base import BaseCommand
from django.utils import timezone
from datetime import timedelta
from apps.tasks.models import Task

class Command(BaseCommand):
    help = "Send due date reminders for tasks approaching deadline"

    def handle(self, *args, **options):
        now = timezone.now()
        soon = now + timedelta(hours=24)
        qs = Task.objects.filter(status__in=["TODO","DOING"], due_date__lte=soon, due_date__gte=now)
        for t in qs:
            self.stdout.write(f"[REMINDER] Task {t.id} '{t.title}' due by {t.due_date}, assignee={t.assignee_id}")
        self.stdout.write(self.style.SUCCESS(f"Processed {qs.count()} reminders"))

tests/conftest.py

import pytest
from django.contrib.auth import get_user_model

@pytest.fixture
def user(db):
    User = get_user_model()
    u = User.objects.create_user(username="tester", password="Test123!", role="member")
    return u

@pytest.fixture
def admin(db):
    User = get_user_model()
    a = User.objects.create_user(username="admin", password="Admin123!", role="admin")
    return a

tests/test_models.py

from apps.tasks.models import Task
from apps.projects.models import Project

def test_create_task(db, user):
    p = Project.objects.create(name="P1", description="d", owner=user)
    t = Task.objects.create(title="T1", project=p, assignee=user)
    assert t.id and t.project_id == p.id

tests/test_services.py

import pytest
from apps.tasks.services import TaskService
from apps.tasks.models import Task
from apps.projects.models import Project

def test_status_transition(db, admin, user):
    p = Project.objects.create(name="P1", description="d", owner=user)
    t = Task.objects.create(title="T", project=p, assignee=user, status="TODO")
    TaskService.change_status(admin, t, "DOING")
    assert Task.objects.get(id=t.id).status == "DOING"
    with pytest.raises(Exception):
        TaskService.change_status(admin, t, "TODO")  # DOING -> TODO allowed, but we test invalid later

tests/test_api.py

from rest_framework.test import APIClient

def test_register_and_login(db):
    c = APIClient()
    r = c.post("/auth/register", {"username":"u1","email":"u1@e.com","password":"User123!"}, format="json")
    assert r.status_code == 201
    r2 = c.post("/auth/login", {"username":"u1","password":"User123!"}, format="json")
    assert r2.status_code == 200 and "access" in r2.json()

Dockerfile

FROM python:3.11-slim

ENV PYTHONDONTWRITEBYTECODE 1
ENV PYTHONUNBUFFERED 1
WORKDIR /app

RUN apt-get update && apt-get install -y build-essential libpq-dev && rm -rf /var/lib/apt/lists/*

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .
ENV DJANGO_SETTINGS_MODULE=config.settings.dev
RUN python manage.py collectstatic --noinput || true

EXPOSE 8000
CMD ["gunicorn", "config.wsgi:application", "-b", "0.0.0.0:8000", "--workers", "3"]

docker-compose.yml

version: "3.9"
services:
  web:
    build: .
    env_file: .env
    ports: ["8000:8000"]
    depends_on: [db, redis]
    volumes: [".:/app"]
  db:
    image: postgres:15
    environment:
      POSTGRES_USER: task
      POSTGRES_PASSWORD: task
      POSTGRES_DB: task
    ports: ["5432:5432"]
  redis:
    image: redis:7
    ports: ["6379:6379"]

.env.example

DJANGO_SETTINGS_MODULE=config.settings.dev
SECRET_KEY=change-me
DEBUG=True
ALLOWED_HOSTS=*
DATABASE_URL=postgres://task:task@db:5432/task
REDIS_URL=redis://redis:6379/1
PAGE_SIZE=20
JWT_ACCESS_MIN=60
JWT_REFRESH_DAYS=7
TIME_ZONE=Asia/Shanghai
FEATURE_RATE_LIMIT=True

运行步骤

• 克隆本项目并进入目录
• 本地运行（SQLite）
  • 创建并填写 .env（不设置 DATABASE_URL 则默认 SQLite）
  • pip install -r requirements.txt
  • python manage.py migrate
  • python manage.py seed
  • python manage.py runserver
• Docker 运行（PostgreSQL/Redis）
  • cp .env.example .env 并根据需要修改
  • docker-compose up --build
  • 初始化数据：docker-compose exec web python manage.py migrate && python manage.py seed
• 提醒任务（可用 cron 触发）
  • python manage.py remind_due

示例 API 调用

• 注册
  • http POST :8000/auth/register username=alice email=alice@ex.com password=Alice123!
• 登录获取 JWT
  • http POST :8000/auth/login username=alice password=Alice123!
• 创建项目
  • http POST :8000/projects/ "Authorization: Bearer " name=Demo description=desc
• 创建任务
  • http POST :8000/tasks/ "Authorization: Bearer " title="TaskA" project=<project_id> assignee=<user_id> priority=高 status=TODO
• 筛选/排序/搜索任务
  • GET /tasks/?status=TODO&priority=高&project=1&ordering=-updated_at&search=关键字&sort_strategy=priority_desc
• 状态流转
  • http POST :8000/tasks/{id}/change_status "Authorization: Bearer " status=DOING
• 任务评论
  • http POST :8000/tasks/{id}/comments "Authorization: Bearer <token)" content="LGTM"
• 文件上传
  • http -f POST :8000/tasks/{id}/attachments "Authorization: Bearer " file@path/to/file.pdf
• 批量导入 CSV
  • http -f POST :8000/tasks/import-csv "Authorization: Bearer " file@tasks.csv project_id==1 default_assignee==2
  • CSV 列：title,description,priority,status,due_date
• Webhook 管理
  • http POST :8000/tasks/webhooks/ "Authorization: Bearer <admin_token>" url=https://example.com/hook event=task.status_changed active:=true secret=abc

OpenAPI

• Schema: GET /api/schema/
• Swagger UI: GET /api/docs/

权限与角色

• admin: 全量权限，管理 Webhook/Label；可见所有项目和任务
• member: 仅访问本人所有的项目与指派给自己的任务；对非本人资源受限
• 自定义权限在 apps/common/permissions.py

架构说明与设计模式映射

• 分层架构
  • views/serializers 作为 Controller/DTO
  • services 处理业务规则（状态流转、导入、提醒、附件/评论）
  • repositories 访问数据（查询封装、缓存点）
• DTO 与验证器
  • serializers 为请求/响应 DTO，validators.CSVImportValidator 分离 CSV 校验
• 策略模式
  • apps/tasks/strategies.py 封装排序策略，通过 sort_strategy 参数切换
• 工厂模式
  • TaskFactory 创建默认任务与初始标签
• 观察者模式
  • observers WebhookNotifier/LoggerNotifier 监听任务状态变化并触发外部回调
• 模板方法
  • TemplateCRUDViewSet 暴露 after_create/after_update/after_destroy 扩展点

可选与扩展建议

• 速率限制：已在 REST_FRAMEWORK 中配置，可通过 FEATURE_RATE_LIMIT 控制开关或自定义 Throttle 类
• 缓存热门查询：在 repositories 中通过 cache 或 cache_page 装饰器增加缓存层，建议对按 project 列表增加键如 tasks:list:project:{id}:{status}:{page}
• 标签层级化：已支持 parent 字段，可实现树形查询
• 任务依赖关系：在 Task 模型中新增 ManyToManyField('self', symmetrical=False, related_name='dependents')，在服务层校验循环依赖
• Webhook 异步化：集成 Celery + Redis，将 Webhook 推送移至异步任务队列，支持重试与退避
• 邮件/IM 提醒：将 remind_due 集成邮件服务或企业微信/Slack Webhook
• 审计日志：新增 AuditLog 模型，信号记录关键动作
• PostgreSQL 特性：JSONB 字段存储自定义属性，GIN 索引提高搜索
• 安全
  • CSRF、CORS、Content Security Policy、文件扩展名白名单、病毒扫描（ClamAV）
  • SECRET_KEY、数据库凭据、Webhook secret 通过环境变量与 KMS/Secret Manager 管理
• 质量与静态检查
  • flake8/black/isort/mypy 集成
  • pre-commit 钩子自动格式化
• CI 示例（GitHub Actions）
  • 安装依赖 -> 迁移 -> pytest --cov -> 构建镜像 -> 推送

CI 工作流示例 .github/workflows/ci.yml（摘录）

name: CI
on: [push, pull_request]
jobs:
  test:
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:15
        env:
          POSTGRES_USER: task
          POSTGRES_PASSWORD: task
          POSTGRES_DB: task
        ports: ["5432:5432"]
        options: >-
          --health-cmd="pg_isready -U task" --health-interval=10s --health-timeout=5s --health-retries=5
    env:
      DATABASE_URL: postgres://task:task@localhost:5432/task
      DJANGO_SETTINGS_MODULE: config.settings.dev
      SECRET_KEY: test
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: "3.11" }
      - run: pip install -r requirements.txt
      - run: python manage.py migrate
      - run: pytest --cov=apps --maxfail=1 -q

常见问题与排错

• 登录失败：确保用户密码符合 Django 密码验证器；seed 数据中的密码 Admin123!/User123!
• 文件上传 413：调整反向代理或 Django FILE_UPLOAD_MAX_MEMORY_SIZE
• Webhook 超时：默认同步 5s 超时，生产建议使用 Celery 异步推送
• 排序字段无效：使用 ordering 参数或 sort_strategy；可在 strategies 中新增类并注册

至此，模板已满足：注册/登录、JWT、角色权限、项目与任务 CRUD、文件上传、评论、状态流转、截止时间提醒、分页、筛选/排序/搜索、CSV 导入、Webhook、结构化日志与错误处理、Swagger 文档、环境与 Docker、测试与种子数据、可扩展性建议。您可直接按运行步骤启动并逐步扩展。需要我为某一模块补全更多细节或加上 Celery/Redis 异步方案，请告诉我。