作者:互联网 时间: 2026-08-19 09:56:55
Django5+Vue3+Docker打造企业OA系统:全栈架构设计与实践的重点在于把前置条件、操作顺序和容易误判的地方分清楚。
企业办公自动化(OA)系统是现代企业数字化转型的核心工具,涵盖流程审批、文档管理、任务协同、人事考勤等模块。本文不堆砌概念,而是从零开始,基于 Django 5.0 Vue 3.4 Docker 构建一套可投产的企业级OA系统后端与前端,并完整交付容器化部署方案。你将收获:

全文代码均经过实测,结构清晰,可直接作为项目脚手架。
┌─────────────────────────────────────────────────┐│Nginx (负载均衡 静态资源)│├─────────────────────────────────────────────────┤│Vue 3 SPA (Pinia Vue Router Axios)│├─────────────────────────────────────────────────┤│Django 5 REST API (JWT RBAC ORM)││- 业务模块:用户/部门/审批/任务/通知 │├─────────────────────────────────────────────────┤│Celery Worker (异步邮件/定时任务)││Redis (缓存 消息队列)││PostgreSQL (主数据库)│└─────────────────────────────────────────────────┘
组件 | 版本 | 关键特性 |
|---|---|---|
Django | 5.0.6 | 支持异步ORM、自动字段类型提示 |
Django REST Framework | 3.15.1 | 视图集、序列化器、权限类 |
Simple JWT | 5.3.1 | Refresh/Access 双令牌机制 |
Vue | 3.4.27 | 响应式Proxy、组合式API |
Vite | 5.2.11 | 极速冷启动、HMR |
Element Plus | 2.7.6 | 企业级组件库 |
PostgreSQL | 16 | 支持JSONB、全文检索 |
Redis | 7.2 | 持久化 订阅发布 |
Docker | 26.0 | BuildKit 加速构建 |
使用 uv 或 pip 创建虚拟环境,安装依赖:
mkdir oa-backend && cd oa-backendpython -m venv venvsource venv/bin/activatepip install django==5.0.6 djangorestframework==3.15.1 djangorestframework-simplejwt==5.3.1 psycopg2-binary redis celery django-cors-headers django-environ
项目结构(核心模块):
代码语言:javascript复制oa_backend/├── oa_backend/│ ├── settings.py# 分层配置(开发/生产)│ ├── urls.py│ └── asgi.py# 支持异步WebSocket├── apps/│ ├── users/ # 用户与权限(自定义User模型)│ ├── departments/ # 部门管理│ ├── approvals/ # 审批流程│ ├── tasks/ # 任务分配│ └── notifications/ # 通知(包含WebSocket)├── core/# 公共组件│ ├── authentication.py# JWT自定义认证类│ ├── permissions.py # RBAC权限类│ ├── pagination.py# 自定义分页│ └── exceptions.py# 全局异常处理器└── celery.py# Celery应用入口
# apps/users/models.pyfrom django.contrib.auth.models import AbstractBaseUser, PermissionsMixinfrom django.db import modelsfrom django.utils import timezonefrom django.utils.translation import gettext_lazy as _class User(AbstractBaseUser, PermissionsMixin):email = models.EmailField(_("email address"), unique=True)full_name = models.CharField(_("full name"), max_length=150)avatar = models.URLField(blank=True, null=True)is_staff = models.BooleanField(default=False)is_active = models.BooleanField(default=True)date_joined = models.DateTimeField(default=timezone.now)# Django 5 新增:使用 Field.choices 更简洁class RoleChoices(models.TextChoices):ADMIN = "ADMIN", _("Admin")MANAGER = "MANAGER", _("Manager")EMPLOYEE = "EMPLOYEE", _("Employee")role = models.CharField(max_length=20,choices=RoleChoices.choices,default=RoleChoices.EMPLOYEE,)# 关联部门(外键)department = models.ForeignKey("departments.Department",on_delete=models.SET_NULL,null=True,blank=True,related_name="members",)USERNAME_FIELD = "email"REQUIRED_FIELDS = ["full_name"]class Meta:db_table = "users"indexes = [models.Index(fields=["email"]),models.Index(fields=["department", "role"]),]def __str__(self):return self.email
注意:Django 5 对 Meta.indexes 支持更丰富的表达式索引,可后续扩展。
修改 settings.py:
# oa_backend/settings.pyfrom datetime import timedeltaREST_FRAMEWORK = {"DEFAULT_AUTHENTICATION_CLASSES": ("core.authentication.CustomJWTAuthentication",# 自定义),"DEFAULT_PERMISSION_CLASSES": ("rest_framework.permissions.IsAuthenticated",),"DEFAULT_PAGINATION_CLASS": "core.pagination.CustomPagination","PAGE_SIZE": 20,"EXCEPTION_HANDLER": "core.exceptions.custom_exception_handler",}SIMPLE_JWT = {"ACCESS_TOKEN_LIFETIME": timedelta(minutes=30),"REFRESH_TOKEN_LIFETIME": timedelta(days=7),"ROTATE_REFRESH_TOKENS": True,"BLACKLIST_AFTER_ROTATION": True,"ALGORITHM": "HS256","SIGNING_KEY": SECRET_KEY,"AUTH_HEADER_TYPES": ("Bearer",),"USER_ID_FIELD": "id","USER_ID_CLAIM": "user_id",}
自定义认证类(支持从Cookie获取Token,前后端分离更安全):
代码语言:javascript复制# core/authentication.pyfrom rest_framework_simplejwt.authentication import JWTAuthenticationfrom django.conf import settingsclass CustomJWTAuthentication(JWTAuthentication):def authenticate(self, request):# 优先从Authorization Header获取header = self.get_header(request)if header is None:# 降级从Cookie获取raw_token = request.COOKIES.get("access_token")else:raw_token = self.get_raw_token(header)if raw_token is None:return Nonevalidated_token = self.get_validated_token(raw_token)return self.get_user(validated_token), validated_token
我们采用 Django 原生 Permission 自定义 Object-level 权限。
代码语言:javascript复制# core/permissions.pyfrom rest_framework.permissions import BasePermissionclass IsAdminOrReadOnly(BasePermission):def has_permission(self, request, view):if request.method in ("GET", "HEAD", "OPTIONS"):return Truereturn request.user and request.user.role == "ADMIN"class IsDepartmentManager(BasePermission):"""部门经理只能操作本部门成员数据"""def has_object_permission(self, request, view, obj):if request.user.role == "ADMIN":return True# 假设 obj 有 department 属性return obj.department == request.user.department and request.user.role == "MANAGER"
在视图集中使用:
代码语言:javascript复制# apps/users/views.pyfrom rest_framework import viewsetsfrom rest_framework.permissions import IsAuthenticatedfrom core.permissions import IsAdminOrReadOnly, IsDepartmentManagerfrom .models import Userfrom .serializers import UserSerializerclass UserViewSet(viewsets.ModelViewSet):queryset = User.objects.select_related("department").all()serializer_class = UserSerializerpermission_classes = [IsAuthenticated, IsAdminOrReadOnly | IsDepartmentManager]def get_queryset(self):# 非管理员只能看到本部门成员if self.request.user.role != "ADMIN":return self.queryset.filter(department=self.request.user.department)return self.queryset
Django 5 的模型字段支持更精确的类型提示,结合 django-stubs 可获得IDE智能提示。序列化器示例:
# apps/users/serializers.pyfrom rest_framework import serializersfrom django.contrib.auth.password_validation import validate_passwordfrom .models import Userclass UserSerializer(serializers.ModelSerializer):# 写入时密码字段不返回password = serializers.CharField(write_only=True, validators=[validate_password])class Meta:model = Userfields = ("id", "email", "full_name", "avatar", "role", "department", "password", "date_joined")read_only_fields = ("id", "date_joined")def create(self, validated_data):password = validated_data.pop("password")user = User(validated_data)user.set_password(password)# 自动hashuser.save()return user
# core/exceptions.pyfrom rest_framework.views import exception_handlerfrom rest_framework.response import Responsefrom rest_framework import statusdef custom_exception_handler(exc, context):response = exception_handler(exc, context)if response is not None:# 统一包装成 {code, message, data}return Response({"code": response.status_code,"message": response.data.get("detail") or str(response.data),"data": None,}, status=response.status_code)return Response({"code": 500,"message": "Internal Server Error","data": None,}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
npm create vite@latest oa-frontend -- --template vue-tscd oa-frontendnpm install vue-router@4 pinia axios element-plus @element-plus/icons-vuenpm install -D @types/node sass
目录结构:
代码语言:javascript复制src/├── api/ # 接口分层│ ├── modules/│ │ ├── user.ts│ │ ├── approval.ts│ │ └── task.ts│ └── request.ts # axios 实例 拦截器├── assets/├── components/# 公共组件├── composables/ # 组合式函数(useAuth, useTable等)├── layouts/├── router/# 路由守卫(权限控制)├── stores/# Pinia modules│ ├── user.ts│ ├── app.ts│ └── notification.ts├── types/ # 全局类型声明├── utils/ # 工具函数├── views/ # 页面视图├── App.vue└── main.ts
// src/api/request.tsimport axios, { AxiosError, InternalAxiosRequestConfig } from 'axios';import { useUserStore } from '@/stores/user';import { ElMessage } from 'element-plus';const request = axios.create({baseURL: import.meta.env.VITE_API_BASE_URL,timeout: 10000,withCredentials: true, // 携带Cookie});// 请求拦截器request.interceptors.request.use((config: InternalAxiosRequestConfig) => {const userStore = useUserStore();if (userStore.accessToken) {config.headers.Authorization = `Bearer ${userStore.accessToken}`;}return config;},(error) => Promise.reject(error));// 响应拦截器 - 自动刷新Tokenlet isRefreshing = false;let failedQueue: any[] = [];request.interceptors.response.use((response) => response,async (error: AxiosError) => {const originalRequest = error.config as InternalAxiosRequestConfig & { _retry?: boolean };if (error.response?.status === 401 && !originalRequest._retry) {if (isRefreshing) {return new Promise((resolve) => {failedQueue.push({ resolve, reject: () => {} });}).then((token) => {originalRequest.headers.Authorization = `Bearer ${token}`;return request(originalRequest);});}originalRequest._retry = true;isRefreshing = true;const userStore = useUserStore();try {const { data } = await request.post('/auth/refresh/', {refresh: userStore.refreshToken,});userStore.setTokens(data.access, data.refresh);// 重试队列failedQueue.forEach((prom) => prom.resolve(data.access));failedQueue = [];originalRequest.headers.Authorization = `Bearer ${data.access}`;return request(originalRequest);} catch (refreshError) {userStore.logout();ElMessage.error('登录已过期,请重新登录');window.location.href = '/login';return Promise.reject(refreshError);} finally {isRefreshing = false;}}return Promise.reject(error);});export default request;
// src/stores/user.tsimport { defineStore } from 'pinia';import { ref, computed } from 'vue';import request from '@/api/request';import type { UserInfo, Role } from '@/types';export const useUserStore = defineStore('user', () => {const userInfo = ref<UserInfo | null>(null);const accessToken = ref<string>('');const refreshToken = ref<string>('');const role = computed(() => userInfo.value?.role || 'EMPLOYEE');const isAdmin = computed(() => role.value === 'ADMIN');const isManager = computed(() => role.value === 'MANAGER');function setTokens(access: string, refresh: string) {accessToken.value = access;refreshToken.value = refresh;localStorage.setItem('refresh_token', refresh);// 可写入Cookie(HttpOnly由后端控制更安全,此处仅做演示)}async function fetchUserInfo() {const { data } = await request.get('/users/me/');userInfo.value = data;return data;}async function login(email: string, password: string) {const { data } = await request.post('/auth/login/', { email, password });setTokens(data.access, data.refresh);await fetchUserInfo();return data;}function logout() {userInfo.value = null;accessToken.value = '';refreshToken.value = '';localStorage.removeItem('refresh_token');// 清理所有状态}return { userInfo, accessToken, refreshToken, role, isAdmin, isManager, setTokens, fetchUserInfo, login, logout };});
// src/router/index.tsimport { createRouter, createWebHistory } from 'vue-router';import { useUserStore } from '@/stores/user';const routes = [{ path: '/login', component: () => import('@/views/Login.vue') },{path: '/',component: () => import('@/layouts/MainLayout.vue'),meta: { requiresAuth: true },children: [{ path: 'dashboard', component: () => import('@/views/Dashboard.vue') },{ path: 'users', component: () => import('@/views/UserList.vue'), meta: { roles: ['ADMIN', 'MANAGER'] } },{ path: 'approvals', component: () => import('@/views/ApprovalList.vue') },],},];const router = createRouter({ history: createWebHistory(), routes });router.beforeEach(async (to, from, next) => {const userStore = useUserStore();const token = userStore.accessToken || localStorage.getItem('refresh_token');if (to.meta.requiresAuth) {if (!token) {return next('/login');}// 尝试获取用户信息,如果未加载if (!userStore.userInfo) {try {await userStore.fetchUserInfo();} catch {userStore.logout();return next('/login');}}// 角色校验if (to.meta.roles) {const roles = to.meta.roles as string[];if (!roles.includes(userStore.role)) {return next('/403');}}next();} else {next();}});export default router;
审批模型使用 Django 的 Choices 和 JSONField 存储动态表单数据:
# apps/approvals/models.pyfrom django.db import modelsfrom django.contrib.postgres.fields import JSONFieldfrom apps.users.models import Userclass Approval(models.Model):class StatusChoices(models.TextChoices):PENDING = "PENDING", "待审批"APPROVED = "APPROVED", "已通过"REJECTED = "REJECTED", "已驳回"CANCELLED = "CANCELLED", "已撤销"title = models.CharField(max_length=200)content = JSONField(default=dict)# 表单字段动态存储applicant = models.ForeignKey(User, on_delete=models.CASCADE, related_name="my_approvals")approver = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, related_name="to_approve")status = models.CharField(max_length=20, choices=StatusChoices.choices, default=StatusChoices.PENDING)created_at = models.DateTimeField(auto_now_add=True)updated_at = models.DateTimeField(auto_now=True)approved_at = models.DateTimeField(null=True, blank=True)comment = models.TextField(blank=True)class Meta:indexes = [models.Index(fields=["status", "approver"]),models.Index(fields=["applicant", "-created_at"]),]
审批视图逻辑(带事务和权限):
代码语言:javascript复制# apps/approvals/views.pyfrom rest_framework import viewsets, statusfrom rest_framework.decorators import actionfrom rest_framework.response import Responsefrom django.db import transactionfrom .models import Approvalfrom .serializers import ApprovalSerializerclass ApprovalViewSet(viewsets.ModelViewSet):serializer_class = ApprovalSerializerpermission_classes = [IsAuthenticated]def get_queryset(self):user = self.request.userif user.role == "ADMIN":return Approval.objects.all()# 申请人或审批人可见return Approval.objects.filter(models.Q(applicant=user) | models.Q(approver=user))@action(detail=True, methods=["post"])@transaction.atomicdef approve(self, request, pk=None):approval = self.get_object()if approval.approver != request.user:return Response({"error": "无权限审批"}, status=status.HTTP_403_FORBIDDEN)if approval.status != "PENDING":return Response({"error": "已处理"}, status=status.HTTP_400_BAD_REQUEST)approval.status = "APPROVED"approval.approved_at = timezone.now()approval.save()# 触发异步通知(Celery)from apps.notifications.tasks import send_approval_notificationsend_approval_notification.delay(approval.id, "APPROVED")return Response({"status": "approved"})
Celery 配置:
代码语言:javascript复制# oa_backend/celery.pyimport osfrom celery import Celeryos.environ.setdefault("DJANGO_SETTINGS_MODULE", "oa_backend.settings")app = Celery("oa_backend")app.config_from_object("django.conf:settings", namespace="CELERY")app.autodiscover_tasks()
settings 添加:
代码语言:javascript复制CELERY_BROKER_URL = os.environ.get("REDIS_URL", "redis://redis:6379/0")CELERY_RESULT_BACKEND = "redis://redis:6379/1"CELERY_ACCEPT_CONTENT = ["json"]CELERY_TASK_SERIALIZER = "json"CELERY_TIMEZONE = "Asia/Shanghai"CELERY_BEAT_SCHEDULE = {"daily-report": {"task": "apps.tasks.tasks.daily_report","schedule": crontab(hour=9, minute=0),},}
异步通知任务:
代码语言:javascript复制# apps/notifications/tasks.pyfrom celery import shared_taskfrom django.core.mail import send_mailfrom django.template.loader import render_to_string@shared_taskdef send_approval_notification(approval_id, status):from .models import Approvalapproval = Approval.objects.select_related("applicant", "approver").get(id=approval_id)subject = f"审批{status}"message = render_to_string("email/approval.html", {"approval": approval, "status": status})send_mail(subject, message, "[email protected]", [approval.applicant.email])
为实现实时消息推送,我们集成 channels(Django 5 支持 ASGI)。
# oa_backend/asgi.pyimport osfrom django.core.asgi import get_asgi_applicationfrom channels.routing import ProtocolTypeRouter, URLRouterfrom channels.auth import AuthMiddlewareStackfrom apps.notifications.routing import websocket_urlpatternsos.environ.setdefault("DJANGO_SETTINGS_MODULE", "oa_backend.settings")application = ProtocolTypeRouter({"http": get_asgi_application(),"websocket": AuthMiddlewareStack(URLRouter(websocket_urlpatterns)),})
消费者(处理连接与消息推送):
代码语言:javascript复制# apps/notifications/consumers.pyimport jsonfrom channels.generic.websocket import AsyncWebsocketConsumerfrom channels.db import database_sync_to_asyncfrom apps.users.models import Userclass NotificationConsumer(AsyncWebsocketConsumer):async def connect(self):self.user = self.scope["user"]if self.user.is_anonymous:await self.close()else: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 receive(self, text_data):# 处理客户端心跳等passasync def notify(self, event):# 发送通知给前端await self.send(text_data=json.dumps({"type": event["type"],"message": event["message"],"data": event.get("data"),}))
# oa-backend/DockerfileFROM python:3.11-slim-bookworm AS builderWORKDIR /appENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1RUN apt-get update && apt-get install -y --no-install-recommends gcc libpq-dev && rm -rf /var/lib/apt/lists/*COPY requirements.txt .RUN pip install --no-cache-dir -r requirements.txt# 生产阶段FROM python:3.11-slim-bookwormWORKDIR /appCOPY --from=builder /usr/local/lib/python3.11/site-packages /usr/local/lib/python3.11/site-packagesCOPY --from=builder /usr/local/bin /usr/local/binCOPY . .RUN useradd -m -u 1000 oa && chown -R oa:oa /appUSER oaEXPOSE 8000CMD ["gunicorn", "oa_backend.wsgi:application", "--bind", "0.0.0.0:8000", "--workers", "4"]
# oa-frontend/DockerfileFROM node:20-alpine AS builderWORKDIR /appCOPY package*.json ./RUN npm ci --only=productionCOPY . .RUN npm run buildFROM nginx:alpineCOPY --from=builder /app/dist /usr/share/nginx/htmlCOPY nginx.conf /etc/nginx/conf.d/default.confEXPOSE 80CMD ["nginx", "-g", "daemon off;"]
# docker-compose.ymlversion: '3.8'services:postgres:image: postgres:16-alpinecontainer_name: oa-postgresenvironment:POSTGRES_DB: oa_dbPOSTGRES_USER: oa_userPOSTGRES_PASSWORD: ${DB_PASSWORD}volumes:- pg_data:/var/lib/postgresql/dataports:- "5432:5432"healthcheck:test: ["CMD-SHELL", "pg_isready -U oa_user"]interval: 10stimeout: 5sretries: 5redis:image: redis:7-alpinecontainer_name: oa-rediscommand: redis-server --appendonly yesvolumes:- redis_data:/dataports:- "6379:6379"backend:build: ./oa-backendcontainer_name: oa-backenddepends_on:postgres:condition: service_healthyredis:condition: service_startedenvironment:- DJANGO_SETTINGS_MODULE=oa_backend.settings.production- DATABASE_URL=postgres://oa_user:${DB_PASSWORD}@postgres:5432/oa_db- REDIS_URL=redis://redis:6379/0volumes:- static_volume:/app/staticfiles- media_volume:/app/mediaports:- "8000:8000"command: >sh -c "python manage.py migrate && python manage.py collectstatic --noinput && gunicorn oa_backend.wsgi:application --bind 0.0.0.0:8000 --workers 4"celery_worker:build: ./oa-backendcontainer_name: oa-celerydepends_on:- redis- postgresenvironment:- DJANGO_SETTINGS_MODULE=oa_backend.settings.production- REDIS_URL=redis://redis:6379/0command: celery -A oa_backend worker -l info --concurrency=4celery_beat:build: ./oa-backendcontainer_name: oa-celery-beatdepends_on:- redis- postgresenvironment:- DJANGO_SETTINGS_MODULE=oa_backend.settings.productioncommand: celery -A oa_backend beat -l infofrontend:build: ./oa-frontendcontainer_name: oa-frontenddepends_on:- backendports:- "80:80"nginx:image: nginx:alpinecontainer_name: oa-nginxvolumes:- ./nginx/nginx.conf:/etc/nginx/nginx.conf:ro- static_volume:/static- media_volume:/mediaports:- "443:443"- "80:80"depends_on:- backend- frontendvolumes:pg_data:redis_data:static_volume:media_volume:
# nginx/nginx.confevents { worker_connections 1024; }http {upstream backend {server backend:8000;}upstream frontend {server frontend:80;}server {listen 80;server_name oa.example.com;# 前端静态资源location / {proxy_pass http://frontend;proxy_set_header Host $host;proxy_set_header X-Real-IP $remote_addr;}# API 反向袋里location /api/ {proxy_pass http://backend;proxy_set_header Host $host;proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;proxy_set_header X-Forwarded-Proto $scheme;}# 后台管理静态文件location /static/ {alias /static/;}location /media/ {alias /media/;}# WebSocket 升级location /ws/ {proxy_pass http://backend;proxy_http_version 1.1;proxy_set_header Upgrade $http_upgrade;proxy_set_header Connection "upgrade";}}}
Django 5 支持 db.models.Index 的 condition 和 include(PostgreSQL 部分索引):
class Approval(models.Model):# ...class Meta:indexes = [models.Index(fields=["status", "approver"], name="idx_approval_status_approver"),# 部分索引:只索引待审批记录models.Index(fields=["created_at"],condition=models.Q(status="PENDING"),name="idx_pending_created",),]
使用 select_related 和 prefetch_related 减少查询次数。
Vite 配置 build.rollupOptions 进行代码分割:
// vite.config.tsexport default defineConfig({build: {rollupOptions: {output: {manualChunks: {'element-plus': ['element-plus'],'vue-ecosystem': ['vue', 'vue-router', 'pinia'],},},},chunkSizeWarningLimit: 1000,},});
集成 django-prometheus 暴露指标,配合 Grafana 可视化。
启动所有服务:
代码语言:javascript复制# 拷贝环境变量cp .env.example .env# 构建并启动docker-compose up -d --build# 初始化管理员docker-compose exec backend python manage.py createsuperuser
滚动更新:
代码语言:javascript复制docker-compose pulldocker-compose up -d --no-deps --build backend celery_worker
本文从技术选型、代码实现到容器化部署,完整展示了基于 Django 5 Vue 3 Docker 的企业OA系统构建过程。核心亮点包括:
Django 5 的 ORM 增强与类型提示支持,提升开发体验JWT RBAC 构建安全可扩展的权限体系Vue 3 组合式 API Pinia 实现可维护的前端状态管理Docker 多阶段构建 docker-compose 实现开发/生产环境一致性Celery WebSocket 满足异步任务与实时通信需求