如何使用Django ORM编写无循环查询以计算员工分段年度病假天数

嘿，我完全懂你不想用循环来处理这种分段年度病假统计的烦恼！咱们直接用Django ORM来搞定，不仅效率高，还能避免手动循环带来的性能问题。

首先先明确下咱们的需求核心：

• 每个员工的第一个统计周期，从他第一条病假记录的创建日期开始，往后推365天；
• 第一个周期结束后，下一个周期从该结束日期之后的第一条病假记录创建日期重新开始，再算365天；
• 每个周期内单独统计病假天数，计数器在新周期启动时重置。

先补充下你没写完的Sick模型关键字段（毕竟要做统计，这些字段肯定少不了）：

from django.db import models
from django.utils.translation import gettext_lazy as _

class Sick(models.Model):
    employee = models.ForeignKey(Employee, on_delete=models.CASCADE, related_name='sick_records')
    created_at = models.DateTimeField(auto_now_add=True, verbose_name=_('创建日期'))
    days = models.IntegerField(_('病假天数'))
    
    class TypeSick(models.TextChoices):
        Initial = 'Initial', _('初始病假')
        Prolongation = 'Prolongation', _('延长病假')
        Accident_Travail = 'Accident Travail', _('工伤')
        Accident_Trajet = 'Accident Trajet', _('通勤事故')
        Congés_Maternités = 'Congés Maternités', _('产假')
        AUTRES = 'Autres', _('其他')
    type = models.CharField(max_length=50, choices=TypeSick.choices, default=TypeSick.Initial)

接下来咱们用Django的窗口函数（Django 2.0+支持）来实现无循环统计，分两步走：

第一步：为每条病假标记所属的统计周期 #

咱们需要给每条病假记录计算出它属于哪个周期的起始日期，这里会用到窗口函数来关联同员工的历史记录：

from datetime import timedelta
from django.db.models import F, Sum, Window, Case, When, BooleanField, DateTimeField
from django.db.models.functions import Lag, Min

# 先给每条病假记录标注所属周期
sick_with_period = Sick.objects.annotate(
    # 获取当前员工上一条病假的创建日期
    prev_created_at=Window(
        expression=Lag('created_at'),
        partition_by=['employee_id'],  # 按员工分组
        order_by=F('created_at').asc()  # 按创建时间排序
    ),
    # 获取当前员工的第一条病假日期
    first_sick_date=Window(
        expression=Min('created_at'),
        partition_by=['employee_id']
    )
).annotate(
    # 判断这条记录是否要开启一个新周期：要么是第一条记录，要么超过了上一个周期的结束时间
    is_new_period=Case(
        When(prev_created_at__isnull=True, then=True),  # 第一条记录，开启新周期
        When(
            created_at__gt=F('first_sick_date') + timedelta(days=365),
            then=True  # 超过第一个周期结束，开启新周期
        ),
        default=False,
        output_field=BooleanField()
    )
).annotate(
    # 确定当前记录的周期起始日期：新周期用自己的创建日期，否则沿用前一条的周期起始
    period_start=Case(
        When(is_new_period=True, then=F('created_at')),
        default=Window(
            expression=Lag('period_start'),
            partition_by=['employee_id'],
            order_by=F('created_at').asc()
        ),
        output_field=DateTimeField()
    )
)

第二步：分组统计每个周期的病假天数 #

现在每条记录都有了所属的周期起始日期，咱们直接按员工和周期起始分组，求和病假天数即可：

# 分组统计每个员工每个周期的总病假天数
period_sick_stats = sick_with_period.values(
    'employee_id', 
    'employee__first_name', 
    'employee__last_name', 
    'period_start'
).annotate(
    total_sick_days=Sum('days'),
    # 计算周期结束日期
    period_end=F('period_start') + timedelta(days=365)
).order_by('employee_id', 'period_start')

使用示例 #

你可以直接遍历这个查询集获取统计结果：

for stat in period_sick_stats:
    print(f"员工: {stat['employee__first_name']} {stat['employee__last_name']}")
    print(f"统计周期: {stat['period_start'].date()} ~ {stat['period_end'].date()}")
    print(f"累计病假天数: {stat['total_sick_days']}\n")

一些细节提示 #

• 如果只需要统计特定类型的病假（比如只算初始病假），可以在初始查询里加过滤：Sick.objects.filter(type=Sick.TypeSick.Initial)
• 关于闰年：如果业务要求按自然年或者366天调整，你可以把timedelta(days=365)换成更精准的日期计算，比如用dateutil.relativedelta
• 确保你的Django版本在2.0以上，窗口函数是从这个版本开始支持的，如果是旧版本，可能需要用原生SQL，但尽量优先用ORM方法

备注：内容来源于stack exchange，提问作者DmitryTok