Python全栈项目--基于Django的项目管理系统

项目概述

项目管理系统是企业日常运营中不可或缺的工具，它能够帮助团队高效地组织、跟踪和管理各类项目任务。本文将介绍如何使用Django框架构建一个功能完善的项目管理系统，涵盖从需求分析到部署的完整开发流程。

技术栈选型

后端技术

Django 4.2+: 成熟的Python Web框架，提供完整的ORM、认证系统和管理后台
Django REST Framework: 构建RESTful API，支持前后端分离
Celery: 异步任务队列，处理邮件通知、报表生成等耗时操作
Redis: 缓存和消息队列
PostgreSQL: 生产环境数据库

前端技术

Vue.js 3: 现代化的前端框架
Element Plus: UI组件库
Axios: HTTP客户端
ECharts: 数据可视化

核心功能模块

1. 用户认证与权限管理

系统需要支持多角色用户体系，包括系统管理员、项目经理、开发人员和访客等角色。使用Django内置的认证系统，结合django-guardian实现对象级权限控制。

from django.contrib.auth.models import AbstractUser from django.db import models class User(AbstractUser): ROLE_CHOICES = ( ('admin', '系统管理员'), ('pm', '项目经理'), ('dev', '开发人员'), ('guest', '访客'), ) role = models.CharField(max_length=20, choices=ROLE_CHOICES) avatar = models.ImageField(upload_to='avatars/', null=True, blank=True) phone = models.CharField(max_length=20, blank=True) department = models.CharField(max_length=100, blank=True) class Meta: db_table = 'users'

2. 项目管理

项目是系统的核心实体，需要记录项目的基本信息、状态、成员和时间线。

class Project(models.Model): STATUS_CHOICES = ( ('planning', '规划中'), ('active', '进行中'), ('suspended', '已暂停'), ('completed', '已完成'), ('cancelled', '已取消'), ) name = models.CharField(max_length=200, verbose_name='项目名称') description = models.TextField(verbose_name='项目描述') status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='planning') owner = models.ForeignKey(User, on_delete=models.PROTECT, related_name='owned_projects') members = models.ManyToManyField(User, related_name='projects', through='ProjectMember') start_date = models.DateField() end_date = models.DateField(null=True, blank=True) budget = models.DecimalField(max_digits=12, decimal_places=2, null=True, blank=True) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: db_table = 'projects' ordering = ['-created_at']

3. 任务管理

任务是项目的最小执行单元，支持任务的创建、分配、状态跟踪和优先级设置。

class Task(models.Model): PRIORITY_CHOICES = ( ('low', '低'), ('medium', '中'), ('high', '高'), ('urgent', '紧急'), ) STATUS_CHOICES = ( ('todo', '待办'), ('in_progress', '进行中'), ('review', '待审核'), ('done', '已完成'), ('blocked', '已阻塞'), ) project = models.ForeignKey(Project, on_delete=models.CASCADE, related_name='tasks') title = models.CharField(max_length=200) description = models.TextField(blank=True) assignee = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, related_name='assigned_tasks') creator = models.ForeignKey(User, on_delete=models.PROTECT, related_name='created_tasks') status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='todo') priority = models.CharField(max_length=20, choices=PRIORITY_CHOICES, default='medium') due_date = models.DateTimeField(null=True, blank=True) estimated_hours = models.DecimalField(max_digits=6, decimal_places=2, null=True, blank=True) actual_hours = models.DecimalField(max_digits=6, decimal_places=2, null=True, blank=True) parent_task = models.ForeignKey('self', on_delete=models.CASCADE, null=True, blank=True, related_name='subtasks') tags = models.ManyToManyField('Tag', blank=True) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: db_table = 'tasks' ordering = ['-priority', '-created_at']

4. 文档管理

项目过程中会产生大量文档，需要提供文档的上传、版本管理和权限控制功能。

class Document(models.Model): project = models.ForeignKey(Project, on_delete=models.CASCADE, related_name='documents') task = models.ForeignKey(Task, on_delete=models.CASCADE, null=True, blank=True, related_name='documents') name = models.CharField(max_length=200) file = models.FileField(upload_to='documents/%Y/%m/%d/') file_size = models.BigIntegerField() file_type = models.CharField(max_length=50) version = models.CharField(max_length=20, default='1.0') uploader = models.ForeignKey(User, on_delete=models.PROTECT) created_at = models.DateTimeField(auto_now_add=True) class Meta: db_table = 'documents'

API设计

使用Django REST Framework构建RESTful API，提供清晰的接口文档和版本管理。

from rest_framework import viewsets, permissions from rest_framework.decorators import action from rest_framework.response import Response class ProjectViewSet(viewsets.ModelViewSet): queryset = Project.objects.all() serializer_class = ProjectSerializer permission_classes = [permissions.IsAuthenticated] def get_queryset(self): user = self.request.user if user.role == 'admin': return Project.objects.all() return user.projects.all() @action(detail=True, methods=['get']) def statistics(self, request, pk=None): project = self.get_object() stats = { 'total_tasks': project.tasks.count(), 'completed_tasks': project.tasks.filter(status='done').count(), 'total_members': project.members.count(), 'progress': self.calculate_progress(project), } return Response(stats) def calculate_progress(self, project): total = project.tasks.count() if total == 0: return 0 completed = project.tasks.filter(status='done').count() return round(completed / total * 100, 2)

前端实现

项目列表页面

使用Vue 3的Composition API和Element Plus构建响应式界面。

<template> <div class="project-list"> <el-card> <template #header> <div class="card-header"> <span>项目列表</span> <el-button type="primary" @click="handleCreate">新建项目</el-button> </div> </template> <el-table :data="projects" style="width: 100%"> <el-table-column prop="name" label="项目名称" /> <el-table-column prop="status" label="状态"> <template #default="{ row }"> <el-tag :type="getStatusType(row.status)"> {{ getStatusText(row.status) }} </el-tag> </template> </el-table-column> <el-table-column prop="owner.username" label="负责人" /> <el-table-column prop="start_date" label="开始日期" /> <el-table-column label="操作"> <template #default="{ row }"> <el-button size="small" @click="handleEdit(row)">编辑</el-button> <el-button size="small" type="danger" @click="handleDelete(row)">删除</el-button> </template> </el-table-column> </el-table> </el-card> </div> </template> <script setup> import { ref, onMounted } from 'vue'; import { getProjects, deleteProject } from '@/api/project'; import { ElMessage } from 'element-plus'; const projects = ref([]); const loadProjects = async () => { try { const response = await getProjects(); projects.value = response.data; } catch (error) { ElMessage.error('加载项目列表失败'); } }; const getStatusType = (status) => { const typeMap = { planning: 'info', active: 'success', suspended: 'warning', completed: '', cancelled: 'danger', }; return typeMap[status]; }; onMounted(() => { loadProjects(); }); </script>

高级特性

1. 实时通知系统

使用WebSocket实现实时消息推送，当任务状态变更或有新评论时及时通知相关用户。

# consumers.py from channels.generic.websocket import AsyncWebsocketConsumer import json class NotificationConsumer(AsyncWebsocketConsumer): async def connect(self): self.user_id = self.scope['user'].id self.group_name = f'user_{self.user_id}' await self.channel_layer.group_add( self.group_name, self.channel_name ) await self.accept() async def disconnect(self, close_code): await self.channel_layer.group_discard( self.group_name, self.channel_name ) async def notification_message(self, event): await self.send(text_data=json.dumps({ 'type': 'notification', 'message': event['message'] }))

2. 数据统计与可视化

提供项目进度、任务分布、成员工作量等多维度的数据分析和图表展示。

from django.db.models import Count, Q from datetime import datetime, timedelta class ProjectStatisticsView(APIView): def get(self, request, pk): project = get_object_or_404(Project, pk=pk) # 任务状态分布 task_status = project.tasks.values('status').annotate(count=Count('id')) # 成员工作量统计 member_workload = project.tasks.values( 'assignee__username' ).annotate( total=Count('id'), completed=Count('id', filter=Q(status='done')) ) # 最近30天任务完成趋势 thirty_days_ago = datetime.now() - timedelta(days=30) daily_completion = project.tasks.filter( status='done', updated_at__gte=thirty_days_ago ).extra( select={'day': 'date(updated_at)'} ).values('day').annotate(count=Count('id')) return Response({ 'task_status': task_status, 'member_workload': member_workload, 'daily_completion': daily_completion, })

3. 导出功能

支持将项目数据导出为Excel或PDF格式，方便生成报告和归档。

from openpyxl import Workbook from django.http import HttpResponse class ProjectExportView(APIView): def get(self, request, pk): project = get_object_or_404(Project, pk=pk) wb = Workbook() ws = wb.active ws.title = "项目任务清单" # 写入表头 headers = ['任务名称', '负责人', '状态', '优先级', '截止日期'] ws.append(headers) # 写入数据 for task in project.tasks.all(): ws.append([ task.title, task.assignee.username if task.assignee else '', task.get_status_display(), task.get_priority_display(), task.due_date.strftime('%Y-%m-%d') if task.due_date else '', ]) response = HttpResponse( content_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet' ) response['Content-Disposition'] = f'attachment; filename={project.name}_tasks.xlsx' wb.save(response) return response

部署方案

Docker容器化部署

使用Docker Compose编排多个服务，简化部署流程。

version: '3.8' services: db: image: postgres:14 environment: POSTGRES_DB: project_management POSTGRES_USER: admin POSTGRES_PASSWORD: password123 volumes: - postgres_data:/var/lib/postgresql/data redis: image: redis:7 command: redis-server --appendonly yes volumes: - redis_data:/data web: build: . command: gunicorn config.wsgi:application --bind 0.0.0.0:8000 volumes: - static_volume:/app/staticfiles - media_volume:/app/media ports: - "8000:8000" depends_on: - db - redis environment: - DATABASE_URL=postgresql://admin:password123@db:5432/project_management - REDIS_URL=redis://redis:6379/0 celery: build: . command: celery -A config worker -l info depends_on: - db - redis nginx: image: nginx:latest ports: - "80:80" volumes: - ./nginx.conf:/etc/nginx/nginx.conf - static_volume:/app/staticfiles - media_volume:/app/media depends_on: - web volumes: postgres_data: redis_data: static_volume: media_volume:

性能优化

1. 数据库查询优化

使用select_related和prefetch_related减少数据库查询次数。

def get_queryset(self): return Project.objects.select_related('owner').prefetch_related( 'members', 'tasks__assignee', 'documents' )

2. 缓存策略

对频繁访问的数据使用Redis缓存。

from django.core.cache import cache def get_project_statistics(project_id): cache_key = f'project_stats_{project_id}' stats = cache.get(cache_key) if stats is None: stats = calculate_statistics(project_id) cache.set(cache_key, stats, 300) # 缓存5分钟 return stats

3. 异步任务处理

将耗时操作放入Celery异步队列。

from celery import shared_task @shared_task def send_task_notification(task_id, user_id): task = Task.objects.get(id=task_id) user = User.objects.get(id=user_id) send_mail( subject=f'新任务分配: {task.title}', message=f'您有新的任务需要处理', from_email='noreply@example.com', recipient_list=[user.email], )

安全考虑

1. 数据验证

在序列化器层面进行严格的数据验证。

from rest_framework import serializers class TaskSerializer(serializers.ModelSerializer): class Meta: model = Task fields = '__all__' def validate_due_date(self, value): if value and value < timezone.now(): raise serializers.ValidationError("截止日期不能早于当前时间") return value

2. 权限控制

实现细粒度的权限控制。

from rest_framework import permissions class IsProjectMember(permissions.BasePermission): def has_object_permission(self, request, view, obj): if request.user.role == 'admin': return True return obj.members.filter(id=request.user.id).exists()

3. API限流

防止恶意请求攻击。

REST_FRAMEWORK = { 'DEFAULT_THROTTLE_CLASSES': [ 'rest_framework.throttling.AnonRateThrottle', 'rest_framework.throttling.UserRateThrottle' ], 'DEFAULT_THROTTLE_RATES': { 'anon': '100/hour', 'user': '1000/hour' } }