如何实现按日期、医疗机构筛选并统计状态为2的保单数量？

需求：统计状态为2的保单（Policy）数量，并按日期和医疗机构（HealthFacility）进行筛选。

涉及的模型类 #

Policy 模型 #

class Policy(core_models.VersionedModel):
    id = models.AutoField(db_column='PolicyID', primary_key=True)
    uuid = models.CharField(db_column='PolicyUUID', max_length=36, default=uuid.uuid4, unique=True)

    stage = models.CharField(db_column='PolicyStage', max_length=1, blank=True, null=True)
    status = models.SmallIntegerField(db_column='PolicyStatus', blank=True, null=True)
    value = models.DecimalField(db_column='PolicyValue', max_digits=18, decimal_places=2, blank=True, null=True)

    family = models.ForeignKey(Family, models.DO_NOTHING, db_column='FamilyID', related_name="policies")
    enroll_date = fields.DateField(db_column='EnrollDate')
    start_date = fields.DateField(db_column='StartDate')
    effective_date = fields.DateField(db_column='EffectiveDate', blank=True, null=True)
    expiry_date = fields.DateField(db_column='ExpiryDate', blank=True, null=True)

    product = models.ForeignKey(Product, models.DO_NOTHING, db_column='ProdID', related_name="policies")
    officer = models.ForeignKey(Officer, models.DO_NOTHING, db_column='OfficerID', blank=True, null=True,
                                related_name="policies")

    offline = models.BooleanField(db_column='isOffline', blank=True, null=True)
    audit_user_id = models.IntegerField(db_column='AuditUserID')

InsureePolicy 模型 #

class InsureePolicy(core_models.VersionedModel):
    id = models.AutoField(db_column='InsureePolicyID', primary_key=True)

    insuree = models.ForeignKey(Insuree, models.DO_NOTHING, db_column='InsureeId', related_name="insuree_policies")
    policy = models.ForeignKey("policy.Policy", models.DO_NOTHING, db_column='PolicyId',
                               related_name="insuree_policies")

    enrollment_date = core.fields.DateField(db_column='EnrollmentDate', blank=True, null=True)
    start_date = core.fields.DateField(db_column='StartDate', blank=True, null=True)
    effective_date = core.fields.DateField(db_column='EffectiveDate', blank=True, null=True)
    expiry_date = core.fields.DateField(db_column='ExpiryDate', blank=True, null=True)

    offline = models.BooleanField(db_column='isOffline', blank=True, null=True)
    audit_user_id = models.IntegerField(db_column='AuditUserID')

Insuree 模型 #

class Insuree(core_models.VersionedModel, core_models.ExtendableModel):
    id = models.AutoField(db_column='InsureeID', primary_key=True)
    uuid = models.CharField(db_column='InsureeUUID', max_length=36, default=uuid.uuid4, unique=True)

    family = models.ForeignKey(Family, models.DO_NOTHING, blank=True, null=True,
                               db_column='FamilyID', related_name="members")
    chf_id = models.CharField(db_column='CHFID', max_length=12, blank=True, null=True)
    last_name = models.CharField(db_column='LastName', max_length=100)
    other_names = models.CharField(db_column='OtherNames', max_length=100)
    gender = models.ForeignKey(Gender, models.DO_NOTHING, db_column='Gender', blank=True, null=True,
                               related_name='insurees')
    dob = core.fields.DateField(db_column='DOB')
    dead = models.BooleanField(db_column='Dead')
    dod = core.fields.DateField(db_column='DOD', null=True,)
    deathReason = models.CharField(db_column='DeathReason', max_length=500, null=True,)
    def age(self, reference_date=None):

    head = models.BooleanField(db_column='IsHead')
    marital = models.CharField(db_column='Marital', max_length=1, blank=True, null=True)

    passport = models.CharField(max_length=25, blank=True, null=True)
    phone = models.CharField(db_column='Phone', max_length=50, blank=True, null=True)
    email = models.CharField(db_column='Email', max_length=100, blank=True, null=True)
    current_address = models.CharField(db_column='CurrentAddress', max_length=200, blank=True, null=True)
    geolocation = models.CharField(db_column='GeoLocation', max_length=250, blank=True, null=True)
    current_village = models.ForeignKey(
        location_models.Location, models.DO_NOTHING, db_column='CurrentVillage', blank=True, null=True)
    photo = models.OneToOneField(InsureePhoto, models.DO_NOTHING,
                              db_column='PhotoID', blank=True, null=True, related_name='+')
    photo_date = core.fields.DateField(db_column='PhotoDate', blank=True, null=True)
    card_issued = models.BooleanField(db_column='CardIssued')
    relationship = models.ForeignKey(
        Relation, models.DO_NOTHING, db_column='Relationship', blank=True, null=True,
        related_name='insurees')
    profession = models.ForeignKey(
        Profession, models.DO_NOTHING, db_column='Profession', blank=True, null=True,
        related_name='insurees')
    education = models.ForeignKey(
        Education, models.DO_NOTHING, db_column='Education', blank=True, null=True,
        related_name='insurees')
    type_of_id = models.ForeignKey(
        IdentificationType, models.DO_NOTHING, db_column='TypeOfId', blank=True, null=True)
    health_facility = models.ForeignKey(
        location_models.HealthFacility, models.DO_NOTHING, db_column='HFID', blank=True, null=True,
        related_name='insurees')

HealthFacility 模型 #

class HealthFacility(core_models.VersionedModel):
    id = models.AutoField(db_column='HfID', primary_key=True)
    uuid = models.CharField(
        db_column='HfUUID', max_length=36, default=uuid.uuid4, unique=True)

    code = models.CharField(db_column='HFCode', max_length=8)
    name = models.CharField(db_column='HFName', max_length=100)
    acc_code = models.CharField(
        db_column='AccCode', max_length=25, blank=True, null=True)
    legal_form = models.ForeignKey(
        HealthFacilityLegalForm, models.DO_NOTHING,
        db_column='LegalForm',
        related_name="health_facilities")
    level = models.CharField(db_column='HFLevel', max_length=1)
    sub_level = models.ForeignKey(
        HealthFacilitySubLevel, models.DO_NOTHING,
        db_column='HFSublevel', blank=True, null=True,
        related_name="health_facilities")
    location = models.ForeignKey(
        Location, models.DO_NOTHING, db_column='LocationId')
    address = models.CharField(
        db_column='HFAddress', max_length=100, blank=True, null=True)
    phone = models.CharField(
        db_column='Phone', max_length=50, blank=True, null=True)
    fax = models.CharField(
        db_column='Fax', max_length=50, blank=True, null=True)
    email = models.CharField(
        db_column='eMail', max_length=50, blank=True, null=True)

    care_type = models.CharField(db_column='HFCareType', max_length=1)

    services_pricelist = models.ForeignKey('medical_pricelist.ServicesPricelist', models.DO_NOTHING,
                                           db_column='PLServiceID', blank=True, null=True,
                                           related_name="health_facilities")
    items_pricelist = models.ForeignKey('medical_pricelist.ItemsPricelist', models.DO_NOTHING, db_column='PLItemID',
                                        blank=True, null=True, related_name="health_facilities")
    offline = models.BooleanField(db_column='OffLine')
    audit_user_id = models.IntegerField(db_column='AuditUserID')

    def __str__(self):
        return self.code + " " + self.name

    @classmethod
    def get_queryset(cls, queryset, user, **kwargs):
        if isinstance(user, ResolveInfo):
            user = user.context.user
        if user.has_perms(LocationConfig.gql_query_health_facilities_perms) and queryset is None:
            queryset = HealthFacility.objects
        else:
            queryset = cls.filter_queryset(queryset)
        if settings.ROW_SECURITY and user.is_anonymous:
            return queryset.filter(id=-1)
        if settings.ROW_SECURITY:
            dist = UserDistrict.get_user_districts(user._u)
            return queryset.filter(
                location_id__in=[l.location_id for l in dist]
            )
        return queryset

现有代码问题分析 #

原cs_in_use_query函数存在以下关键问题：

1. 使用了Policy模型不存在的validity_from/validity_to字段进行日期筛选
2. dict1/dict2/dict3为空字典，未传递任何筛选条件，导致后续查询无效
3. 未建立Policy与HealthFacility的关联关系，无法实现医疗机构筛选
4. 仅查询了ID列表，未完成统计逻辑，返回结果也未包含统计值

修正后的实现逻辑 #

优化后的查询函数 #

from django.db.models import Count
from datetime import datetime

def cs_in_use_query(user, **kwargs):
    date_from = kwargs.get("date_from")
    date_to = kwargs.get("date_to")
    hflocation = kwargs.get("hflocation")  # 医疗机构编码
    hfid = kwargs.get("hfid")  # 医疗机构ID

    # 日期格式转换为Date对象
    format = "%Y-%m-%d"
    date_from_obj = datetime.strptime(date_from, format).date() if date_from else None
    date_to_obj = datetime.strptime(date_to, format).date() if date_to else None

    # 初始化查询集：筛选状态为2的保单
    queryset = Policy.objects.filter(status=2)

    # 日期范围筛选（按生效日期，可根据业务替换为enroll_date/start_date）
    if date_from_obj and date_to_obj:
        queryset = queryset.filter(
            effective_date__gte=date_from_obj,
            effective_date__lte=date_to_obj
        )

    # 医疗机构筛选：通过关联表链式过滤
    if hflocation or hfid:
        hf_filters = {}
        if hfid:
            hf_filters['id'] = hfid
        elif hflocation and hflocation != "0":
            hf_filters['code'] = hflocation
        
        # 关联路径：Policy → InsureePolicy → Insuree → HealthFacility
        queryset = queryset.filter(
            insuree_policies__insuree__health_facility__in=HealthFacility.objects.filter(**hf_filters)
        ).distinct()  # 去重，避免同一保单被多个被保险人关联导致重复统计

    # 统计符合条件的保单数量
    policy_count = queryset.count()

    # 构造返回结果
    result = {
        "dateFrom": date_from,
        "dateTo": date_to,
        "policyCount": policy_count,
    }

    # 添加医疗机构详情（如果有筛选条件）
    if hflocation or hfid:
        hf_obj = HealthFacility.objects.filter(**hf_filters).first()
        if hf_obj:
            result["healthFacility"] = {
                "id": hf_obj.id,
                "code": hf_obj.code,
                "name": hf_obj.name
            }

    return result

关键说明 #

1. 关联查询优化：使用Django ORM链式关联查询，一次性完成所有筛选逻辑，避免多次查询数据库，提升性能
2. 去重处理：通过distinct()确保同一保单不会因关联多个被保险人而被重复统计
3. 日期字段灵活性：可根据业务需求将effective_date替换为enroll_date或start_date
4. 权限兼容：若需要行级权限控制，可复用HealthFacility.get_queryset中的逻辑对查询集进行二次过滤

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