def update_revisions()

in backend/code_review_backend/issues/migrations/0008_revision_base_head_references.py [0:0]

• 27 lines of code
• 3 McCabe index (conditional complexity)

def update_revisions(apps, schema_editor):
    """
    Detect the head repository depending on the current data:
     - all revisions without any diff use their current head repo
     - all revisions with a only diffs on autoland use it as their head repo
    """
    Repository = apps.get_model("issues", "Repository")
    Revision = apps.get_model("issues", "Revision")
    Diff = apps.get_model("issues", "Diff")

    # No need to run the following lines if the db is empty (e.g. during tests)
    if not Revision.objects.exists():
        return

    # all revisions without any diff get their head repo as revision.repository_id
    Revision.objects.filter(diffs__isnull=True).update(
        head_repository_id=F("base_repository_id")
    )

    # all revisions with only diff on autoland use it as their head repo
    autoland = Repository.objects.get(slug="autoland")
    others = Repository.objects.exclude(id=autoland.id)
    Revision.objects.annotate(
        other_diffs=Count("diffs", filter=Q(diffs__repository__in=others))
    ).filter(other_diffs=0, diffs__repository_id=autoland.id).update(
        head_repository_id=autoland.id
    )

    # all revisions without any diff excluding diffs with autoland repo
    # get their head repo as diff.repository_id
    rev_repos = (
        Diff.objects.filter(revision_id=OuterRef("id"))
        .exclude(repository_id=autoland.id)
        .values("repository_id")[:1]
    )
    Revision.objects.filter(head_repository_id__isnull=True).update(
        head_repository_id=Subquery(rev_repos)
    )

    # Check no revisions miss their head repos
    assert not Revision.objects.filter(
        head_repository__isnull=True
    ).exists(), "Some revisions miss their head repos"