优化Django中相似药物SQL查询性能的技术咨询

我来帮你解决这个相似药物查询的性能问题——你的核心需求是找到和目标药物**EPC类别集合完全一致（不多不少）**的药物，当前遍历所有药物逐一比对的方式在数据量上升后确实会变得很慢。下面给你几个高效的实现思路和代码示例：

先回顾你的现有实现 #

views.py #

class GetSimilarDrugs(APIView):
    def get(self, request, format=None):
        get_req = request.GET.get('drugid', '')
        simi_list = []
        comp_class = DrugBankDrugEPClass.objects.filter(drug_bank_id = get_req).values_list('epc_id', flat=True).distinct()
        for drg_id in DrugBankDrugEPClass.objects.values_list('drug_bank_id', flat = True).distinct():
            classtocomp = DrugBankDrugEPClass.objects.filter(drug_bank_id = str(drg_id)).values_list('epc_id', flat=True).distinct()
            complist = list(comp_class)
            tolist = list(classtocomp)
            if complist == tolist:
                simi_list.append(drg_id)
        return Response({'result':simi_list})

models.py #

class DrugBankDrugEPClass(models.Model):
    drug_bank = models.ForeignKey(DrugBankDrugs, on_delete=models.CASCADE)
    epc = models.ForeignKey(DrugBankEPClass, on_delete=models.CASCADE)

测试SQL数据 #

id | drug_bank_id | epc_id |
+------+--------------+--------+
| 1 | DB12789 | 1 |
| 2 | DB12788 | 2 |
| 3 | DB00596 | 3 |
| 4 | DB09161 | 4 |
| 5 | DB01178 | 5 |
| 6 | DB01177 | 6 |
| 7 | DB01177 | 6 |
| 8 | DB01174 | 7 |
| 9 | DB01175 | 8 |
| 10 | DB01172 | 9 |
| 11 | DB01173 | 10 |
| 12 | DB12257 | 11 |
| 13 | DB08167 | 12 |
| 14 | DB01551 | 13 |
| 15 | DB01006 | 14 |
| 16 | DB01007 | 15 |
| 17 | DB01007 | 16 |
| 18 | DB01004 | 17 |
| 19 | DB01004 | 18 |
| 20 | DB01004 | 17 |
| 21 | DB01004 | 18 |
| 22 | DB01004 | 19 |
| 23 | DB00570 | 20 |
| 24 | DB01008 | 21 |
| 25 | DB00572 | 22 |
| 26 | DB00575 | 7 |
| 27 | DB00577 | 23 |
| 28 | DB00577 | 24 |
| 29 | DB00577 | 25 |
| 30 | DB00576 | 26 |
| 31 | DB00751 | 27 |
| 32 | DB00751 | 28 |
| 33 | DB00750 | 29 |
| 34 | DB00753 | 30 |
| 35 | DB00752 | 31 |
| 36 | DB00755 | 32 |
| 37 | DB00755 | 32 |
| 38 | DB00757 | 33 |
| 39 | DB00756 | 34 |
| 40 | DB00759 | 35 |
| 41 | DB00759 | 36 |
| 42 | DB00759 | 36 |

优化方案 #

方案1：数据库聚合+子查询（通用SQL兼容） #

核心思路是把比对逻辑放到数据库层面完成，避免Python循环遍历所有药物：

1. 先获取目标药物的EPC类别数量和集合
2. 筛选出EPC数量和目标一致的候选药物
3. 排除候选药物中包含目标集合外类别的药物

from django.db.models import Count, Q

class GetSimilarDrugs(APIView):
    def get(self, request, format=None):
        drug_id = request.GET.get('drugid', '')
        if not drug_id:
            return Response({'result': []})
        
        # 获取目标药物的EPC集合和数量
        target_epcs = set(DrugBankDrugEPClass.objects.filter(drug_bank_id=drug_id)
                          .values_list('epc_id', flat=True).distinct())
        target_epc_count = len(target_epcs)
        
        if target_epc_count == 0:
            # 处理无EPC类别的情况：返回所有无EPC的药物
            drugs_with_epc = set(DrugBankDrugEPClass.objects.values_list('drug_bank_id', flat=True))
            all_drugs = DrugBankDrugs.objects.values_list('drug_bank_id', flat=True)
            result = [d for d in all_drugs if d not in drugs_with_epc]
            return Response({'result': result})
        
        # 第一步：筛选EPC数量和目标完全一致的药物
        candidates = DrugBankDrugEPClass.objects.values('drug_bank_id')\
            .annotate(epc_count=Count('epc_id', distinct=True))\
            .filter(epc_count=target_epc_count)\
            .values_list('drug_bank_id', flat=True)
        
        # 第二步：排除那些包含目标集合外EPC的药物
        excluded_drugs = DrugBankDrugEPClass.objects.filter(
            drug_bank_id__in=candidates,
            epc_id__not_in=target_epcs
        ).values_list('drug_bank_id', flat=True).distinct()
        
        # 最终结果：候选药物 - 被排除的药物
        simi_list = [d for d in candidates if d not in excluded_drugs]
        
        return Response({'result': simi_list})

方案2：PostgreSQL数组匹配（如果用PostgreSQL） #

如果你的数据库是PostgreSQL，可以利用它的数组聚合功能，直接在数据库层面完成集合比对，效率更高：

from django.db.models import Func
from django.contrib.postgres.fields import ArrayField

class ArrayAggDistinct(Func):
    function = 'ARRAY_AGG'
    template = "%(function)s(%(expressions)s ORDER BY %(expressions)s) FILTER (WHERE %(expressions)s IS NOT NULL)"
    output_field = ArrayField(base_field=Func('epc_id', function='CAST', template='%(function)s(%(expressions)s AS INTEGER)'))

class GetSimilarDrugs(APIView):
    def get(self, request, format=None):
        drug_id = request.GET.get('drugid', '')
        if not drug_id:
            return Response({'result': []})
        
        # 获取目标药物的排序后EPC数组（去重）
        target_epc_array = DrugBankDrugEPClass.objects.filter(drug_bank_id=drug_id)\
            .annotate(epcs=ArrayAggDistinct('epc_id'))\
            .values_list('epcs', flat=True).first()
        
        if not target_epc_array:
            # 处理无EPC的情况
            drugs_with_epc = set(DrugBankDrugEPClass.objects.values_list('drug_bank_id', flat=True))
            all_drugs = DrugBankDrugs.objects.values_list('drug_bank_id', flat=True)
            result = [d for d in all_drugs if d not in drugs_with_epc]
            return Response({'result': result})
        
        # 直接匹配数组完全相等的药物
        simi_list = DrugBankDrugEPClass.objects.values('drug_bank_id')\
            .annotate(epcs=ArrayAggDistinct('epc_id'))\
            .filter(epcs=target_epc_array)\
            .values_list('drug_bank_id', flat=True).distinct()
        
        return Response({'result': list(simi_list)})

方案3：预计算EPC集合哈希值（高频查询场景） #

如果查询非常频繁，可以给药物添加一个哈希字段，存储其EPC集合的哈希值，查询时直接比对哈希即可：

1. 先修改models.py：

import hashlib

class DrugBankDrugs(models.Model):
    # 原有字段...
    epc_set_hash = models.CharField(max_length=64, blank=True, null=True)
    
    def update_epc_hash(self):
        # 生成排序后的EPC字符串，再计算哈希值
        epcs = sorted(DrugBankDrugEPClass.objects.filter(drug_bank=self)
                     .values_list('epc_id', flat=True).distinct())
        epc_str = ','.join(map(str, epcs))
        self.epc_set_hash = hashlib.sha256(epc_str.encode()).hexdigest()
        self.save()

2. 然后修改views.py：

class GetSimilarDrugs(APIView):
    def get(self, request, format=None):
        drug_id = request.GET.get('drugid', '')
        if not drug_id:
            return Response({'result': []})
        
        try:
            target_drug = DrugBankDrugs.objects.get(drug_bank_id=drug_id)
        except DrugBankDrugs.DoesNotExist:
            return Response({'result': []})
        
        # 确保目标药物的哈希值是最新的
        target_drug.update_epc_hash()
        
        # 查询所有哈希值匹配的药物
        simi_list = DrugBankDrugs.objects.filter(epc_set_hash=target_drug.epc_set_hash)\
            .values_list('drug_bank_id', flat=True)
        
        return Response({'result': list(simi_list)})

注意：需要在EPC类别更新时同步更新哈希值，可以用Django信号或者在DrugBankDrugEPClass的save方法中触发。

额外性能优化建议 #

• 给DrugBankDrugEPClass添加联合索引，加速查询：

  class DrugBankDrugEPClass(models.Model):
      drug_bank = models.ForeignKey(DrugBankDrugs, on_delete=models.CASCADE)
      epc = models.ForeignKey(DrugBankEPClass, on_delete=models.CASCADE)
      
      class Meta:
          indexes = [
              models.Index(fields=['drug_bank_id', 'epc_id']),
              models.Index(fields=['epc_id', 'drug_bank_id']),
          ]
  

• 尽量避免在Python中处理大量数据，把筛选逻辑都放到数据库层面完成，减少数据传输和内存占用。

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