Django REST Framework HRMS休假管理：规则、审批与邮件实现咨询

一、模型结构优化 #

你当前的LeaveType模型存在逻辑错误：类定义阶段的if判断无法根据实例的leave_type动态生成规则，需要将休假规则字段化，并新增LeaveApplication模型记录员工的具体申请。

from django.db import models
from django.utils import timezone

class LeaveType(models.Model):
    LEAVE_TYPE_CHOICES = (
        ('CL', '事假'),
        ('SL', '病假'),
        ('ML', '医疗假'),
        ('COMP_OFF', '调休'),
        ('LOP', '事假（L.O.P）')
    )
    LEAVE_DURATION_CHOICES = (
        ('FULL', '全天'),
        ('FORE_NOON', '上午'),
        ('AFTER_NOON', '下午')
    )
    STATUS_CHOICES = (
        ('APPROVED', '已批准'),
        ('REJECTED', '已拒绝'),
        ('PENDING', '待审批')
    )

    leave_type = models.CharField(max_length=20, choices=LEAVE_TYPE_CHOICES)
    annual_quota = models.IntegerField(default=0)       # 年度总天数
    monthly_limit = models.IntegerField(default=0)      # 每月申请上限
    max_carry_over = models.IntegerField(default=0)     # 最大结转天数
    quarterly_limit = models.IntegerField(default=0)    # 季度申请上限

    class Meta:
        verbose_name = "休假类型"
        verbose_name_plural = "休假类型"

class LeaveApplication(models.Model):
    employee = models.ForeignKey('employees.Employee', on_delete=models.CASCADE)
    tl = models.ForeignKey('employees.Employee', on_delete=models.CASCADE, related_name='approved_leaves')
    leave_type = models.ForeignKey(LeaveType, on_delete=models.CASCADE)
    start_date = models.DateField()
    end_date = models.DateField(null=True, blank=True)
    duration_type = models.CharField(max_length=20, choices=LeaveType.LEAVE_DURATION_CHOICES, default='FULL')
    status = models.CharField(max_length=20, choices=LeaveType.STATUS_CHOICES, default='PENDING')
    applied_at = models.DateTimeField(auto_now_add=True)
    approved_at = models.DateTimeField(null=True, blank=True)
    rejection_reason = models.TextField(null=True, blank=True)

    class Meta:
        verbose_name = "休假申请"
        verbose_name_plural = "休假申请"

二、业务逻辑分层 #

1. Serializer层：负责数据校验 #

仅处理数据格式、规则合法性校验，不涉及业务流程：

from rest_framework import serializers
from .models import LeaveApplication, LeaveType
from django.utils import timezone

class LeaveApplicationSerializer(serializers.ModelSerializer):
    employee_name = serializers.ReadOnlyField(source='employee.full_name')
    tl_name = serializers.ReadOnlyField(source='tl.full_name')

    class Meta:
        model = LeaveApplication
        fields = ['id', 'employee', 'tl', 'leave_type', 'start_date', 'end_date', 'duration_type', 'status', 'applied_at', 'employee_name', 'tl_name']
        read_only_fields = ['status', 'applied_at']

    def validate(self, data):
        # 半天假校验：无需填写结束日期
        if data['duration_type'] != 'FULL' and data.get('end_date'):
            raise serializers.ValidationError("半天假无需填写结束日期")
        
        leave_type = data['leave_type']
        employee = data['employee']
        used_days = self._calculate_used_leave_days(employee, leave_type)
        request_days = self._calculate_request_days(data)

        # CL规则校验
        if leave_type.leave_type == 'CL':
            if used_days + request_days > leave_type.annual_quota:
                raise serializers.ValidationError(f"事假年度额度仅剩{leave_type.annual_quota - used_days}天，无法申请本次休假")
            monthly_used = self._calculate_monthly_used(employee, leave_type)
            if monthly_used + request_days > leave_type.monthly_limit:
                raise serializers.ValidationError(f"本月事假已申请{monthly_used}天，最多可申请{leave_type.monthly_limit}天")
        
        # SL规则校验
        elif leave_type.leave_type == 'SL':
            quarterly_used = self._calculate_quarterly_used(employee, leave_type)
            if quarterly_used + request_days > leave_type.quarterly_limit:
                raise serializers.ValidationError(f"本季度病假已申请{quarterly_used}天，最多可申请{leave_type.quarterly_limit}天")
            if used_days + request_days > leave_type.annual_quota:
                raise serializers.ValidationError(f"病假年度额度仅剩{leave_type.annual_quota - used_days}天，无法申请本次休假")
        
        return data

    def _calculate_request_days(self, data):
        if data['duration_type'] != 'FULL':
            return 0.5
        return 1 if not data.get('end_date') else (data['end_date'] - data['start_date']).days + 1

    def _calculate_used_leave_days(self, employee, leave_type):
        approved_leaves = LeaveApplication.objects.filter(
            employee=employee, leave_type=leave_type, status='APPROVED', start_date__year=timezone.now().year
        )
        total_days = 0
        for leave in approved_leaves:
            total_days += 0.5 if leave.duration_type != 'FULL' else (1 if not leave.end_date else (leave.end_date - leave.start_date).days +1)
        return total_days

    def _calculate_monthly_used(self, employee, leave_type):
        current_month = timezone.now().month
        leaves = LeaveApplication.objects.filter(
            employee=employee, leave_type=leave_type, status='APPROVED', start_date__month=current_month, start_date__year=timezone.now().year
        )
        return sum(0.5 if l.duration_type != 'FULL' else 1 for l in leaves)

    def _calculate_quarterly_used(self, employee, leave_type):
        current_quarter = (timezone.now().month -1)//3 +1
        start_month = (current_quarter -1)*3 +1
        end_month = current_quarter*3
        leaves = LeaveApplication.objects.filter(
            employee=employee, leave_type=leave_type, status='APPROVED', 
            start_date__month__gte=start_month, start_date__month__lte=end_month, start_date__year=timezone.now().year
        )
        return sum(0.5 if l.duration_type != 'FULL' else 1 for l in leaves)

2. View层：处理业务流程 #

负责申请提交、审批触发、权限控制等流程，复杂逻辑可抽离到Service层：

from rest_framework import generics, permissions, status
from rest_framework.response import Response
from .models import LeaveApplication
from .serializers import LeaveApplicationSerializer
from .services import send_leave_application_notification, send_leave_status_update_notification
from django.utils import timezone

class LeaveApplicationCreateView(generics.CreateAPIView):
    serializer_class = LeaveApplicationSerializer
    permission_classes = [permissions.IsAuthenticated]

    def perform_create(self, serializer):
        employee = self.request.user.employee
        application = serializer.save(employee=employee, tl=employee.team_leader)
        send_leave_application_notification(application)

class LeaveApprovalView(generics.UpdateAPIView):
    queryset = LeaveApplication.objects.all()
    serializer_class = LeaveApplicationSerializer
    permission_classes = [permissions.IsAuthenticated]

    def update(self, request, *args, **kwargs):
        instance = self.get_object()
        if request.user.employee != instance.tl:
            return Response({"detail": "无审批权限"}, status=status.HTTP_403_FORBIDDEN)
        
        instance.status = request.data.get('status')
        instance.approved_at = timezone.now()
        instance.rejection_reason = request.data.get('rejection_reason', '')
        instance.save()
        
        send_leave_status_update_notification(instance)
        return Response(self.get_serializer(instance).data)

class LeaveApplicationListView(generics.ListAPIView):
    serializer_class = LeaveApplicationSerializer
    permission_classes = [permissions.IsAuthenticated]

    def get_queryset(self):
        user = self.request.user.employee
        if user.role in ['HR', 'MANAGER']:
            return LeaveApplication.objects.all()
        elif user.role == 'TL':
            return LeaveApplication.objects.filter(tl=user)
        else:
            return LeaveApplication.objects.filter(employee=user)

3. Service层：抽离通用逻辑 #

将邮件发送、复杂计算等逻辑独立，让View更简洁：

from django.core.mail import send_mail
from django.conf import settings
from .models import LeaveApplication
from employees.models import Employee

def send_leave_application_notification(application):
    subject = f"新的休假申请待审批 - {application.employee.full_name}"
    message = f"""
    员工{application.employee.full_name}提交了休假申请：
    休假类型：{application.leave_type.get_leave_type_display()}
    休假时间：{application.start_date} {'至' + str(application.end_date) if application.end_date else ''}
    休假时长：{application.get_duration_type_display()}
    请及时登录系统审批。
    """
    send_mail(subject, message, settings.DEFAULT_FROM_EMAIL, [application.tl.email], fail_silently=False)

def send_leave_status_update_notification(application):
    # 通知员工
    employee_subject = f"休假申请状态更新 - {application.get_status_display()}"
    employee_message = f"""
    您的休假申请已{application.get_status_display()}：
    休假类型：{application.leave_type.get_leave_type_display()}
    休假时间：{application.start_date} {'至' + str(application.end_date) if application.end_date else ''}
    {'拒绝理由：' + application.rejection_reason if application.status == 'REJECTED' else ''}
    """
    send_mail(employee_subject, employee_message, settings.DEFAULT_FROM_EMAIL, [application.employee.email], fail_silently=False)
    
    # 通知HR
    hr_subject = f"休假申请状态更新 - {application.employee.full_name}"
    hr_message = f"""
    员工{application.employee.full_name}的休假申请已{application.get_status_display()}：
    休假类型：{application.leave_type.get_leave_type_display()}
    审批人：{application.tl.full_name}
    """
    hr_emails = [emp.email for emp in Employee.objects.filter(role='HR')]
    send_mail(hr_subject, hr_message, settings.DEFAULT_FROM_EMAIL, hr_emails, fail_silently=False)

三、邮件通知配置 #

在settings.py中配置SMTP服务：

EMAIL_BACKEND = 'django.core.mail.backends.smtp.EmailBackend'
EMAIL_HOST = 'smtp.example.com'  # 如smtp.gmail.com
EMAIL_PORT = 587
EMAIL_USE_TLS = True
EMAIL_HOST_USER = 'your-email@example.com'
EMAIL_HOST_PASSWORD = 'your-authorization-code'  # 第三方邮箱用授权码
DEFAULT_FROM_EMAIL = 'your-email@example.com'

四、休假结转逻辑处理 #

针对CL的结转需求，可通过定时任务（如django-crontab或Celery）在每年年初执行：

from django.utils import timezone
from .models import LeaveApplication, LeaveType
from employees.models import Employee

def carry_over_cl_leaves():
    current_year = timezone.now().year
    last_year = current_year -1
    cl_type = LeaveType.objects.get(leave_type='CL')
    
    for employee in Employee.objects.all():
        approved_leaves = LeaveApplication.objects.filter(
            employee=employee, leave_type=cl_type, status='APPROVED', start_date__year=last_year
        )
        used_days = sum(0.5 if l.duration_type != 'FULL' else 1 for l in approved_leaves)
        remaining = cl_type.annual_quota - used_days
        employee.cl_carry_over = min(remaining, cl_type.max_carry_over)
        employee.save()

内容的提问来源于stack exchange，提问作者janezio