Django自关联ManyToMany字段批量创建与查询问题求助

模型定义 #

SoftwareVersion(models.Model):
    id = models.AutoField(
        primary_key=True,
        db_index=True
    )

    ... Some other fields ...

    incompatibilities= models.ManyToManyField(
        "self",
        symmetrical=True,
        blank=True,
        default=None,
        through="Incompatibilities"
    )

Incompatibilities(models.Model):
    id = models.AutoField(
        primary_key=True,
        db_index=True
    )

    softwareversion_a = models.ForeignKey(
        "SoftwareVersion",
        models.CASCADE,
        db_index=True,
        db_column='softwareversion_a ',
        related_name='softwareversion_a',
        verbose_name="software version a",
    )

    softwareversion_b = models.ForeignKey(
        "SoftwareVersion",
        models.CASCADE,
        db_index=True,
        db_column='softwareversion_b',
        related_name='softwareversion_b',
        verbose_name="softwareversion_b",
    )

    status = models.BooleanField(
        verbose_name='Status',
        default=False
    )

    class Meta:
        unique_together = (('softwareversion_a', 'softwareversion_b'),)

问题一：批量创建重复数据 #

问题描述 #

在SoftwareVersion的save方法中用bulk_create批量生成关联记录时，即使设置了ignore_conflicts=True，仍会生成违反unique_together约束的重复记录；循环创建性能又太差。

现有代码 #

# Inside SoftwareVersion Model class
def save(self, force_insert=False, force_update=False, using=None, update_fields=None) -> None:
    save = super().save(force_insert, force_update, using, update_fields)
    
    Incompatibilities.objects.bulk_create(
       (Incompatibilities(
       softwareversion_a=self,
       softwareversion_b=software_version,
       status=False
            ) for software_version in SoftwareVersion.objects.exclude(self)),
            ignore_conflicts=True,
            batch_size=1000
        )

    return save

解决方案 #

核心思路是先查询已存在的关联对，排除后再批量创建，避免生成重复记录：

1. 修正exclude(self)的写法（原写法无法正确排除自身，改用主键排除）
2. 查询所有已存在的关联对，包含对称的(a,b)和(b,a)组合
3. 生成新记录时跳过已存在的组合

修改后的save方法：

from django.db.models import Q

def save(self, force_insert=False, force_update=False, using=None, update_fields=None) -> None:
    super().save(force_insert, force_update, using, update_fields)
    
    # 获取所有已存在的关联对（包含对称组合）
    existing_pairs = Incompatibilities.objects.filter(
        Q(softwareversion_a=self) | Q(softwareversion_b=self)
    ).values_list('softwareversion_a_id', 'softwareversion_b_id')
    
    existing_set = set()
    for a_id, b_id in existing_pairs:
        existing_set.add((a_id, b_id))
        existing_set.add((b_id, a_id))
    
    # 生成未存在的新记录
    new_records = []
    for sv in SoftwareVersion.objects.exclude(pk=self.pk):
        current_pair = (self.pk, sv.pk)
        if current_pair not in existing_set:
            new_records.append(Incompatibilities(
                softwareversion_a=self,
                softwareversion_b=sv,
                status=False
            ))
    
    # 批量插入（仅当有新记录时执行）
    if new_records:
        Incompatibilities.objects.bulk_create(
            new_records,
            ignore_conflicts=True,
            batch_size=1000
        )

额外说明：

• ignore_conflicts=True仅在数据库支持时生效（如PostgreSQL 9.5+、MySQL 8.0.19+），作为最后一层保障
• 提前排除已存在的记录，能大幅减少无效的数据库写入请求，进一步提升性能

问题二：兼容性记录查询不全 #

问题描述 #

查询某个SoftwareVersion的兼容性记录时，仅查询softwareversion_a指向该版本的记录会遗漏softwareversion_b指向它的记录；使用多对多关联的incompatibilities.all()结果相同；尝试Q查询后无法按关联字段正常排序。

解决方案 #

1. 完整查询所有关联记录

用Q查询同时匹配softwareversion_a和softwareversion_b即可获取完整记录：

from django.db.models import Q

current_version = SoftwareVersion.objects.get(pk=your_version_id)
full_incompatibilities = Incompatibilities.objects.filter(
    Q(softwareversion_a=current_version) | Q(softwareversion_b=current_version)
)

2. 按关联版本排序

通过annotate结合Case/When，统一提取关联的版本ID，再按该字段排序：

from django.db.models import Case, When, IntegerField

current_version = SoftwareVersion.objects.get(pk=your_version_id)

sorted_incompatibilities = Incompatibilities.objects.filter(
    Q(softwareversion_a=current_version) | Q(softwareversion_b=current_version)
).annotate(
    # 统一提取关联的版本ID：如果当前版本是a，则取b的ID；否则取a的ID
    related_version_id=Case(
        When(softwareversion_a=current_version, then='softwareversion_b_id'),
        default='softwareversion_a_id',
        output_field=IntegerField()
    )
).order_by('related_version_id')  # 按关联版本ID排序，可改为'-related_version_id'实现降序

3. 封装成模型方法（可选）

为了方便调用，可以在SoftwareVersion模型中封装查询方法：

class SoftwareVersion(models.Model):
    # ... 其他字段和方法 ...
    
    def get_all_incompatibilities(self, sorted=True):
        qs = Incompatibilities.objects.filter(
            Q(softwareversion_a=self) | Q(softwareversion_b=self)
        )
        if sorted:
            qs = qs.annotate(
                related_version_id=Case(
                    When(softwareversion_a=self, then='softwareversion_b_id'),
                    default='softwareversion_a_id',
                    output_field=IntegerField()
                )
            ).order_by('related_version_id')
        return qs

调用示例：

current_version.get_all_incompatibilities()  # 已排序的完整记录
current_version.get_all_incompatibilities(sorted=False)  # 未排序的完整记录

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