def cleanup_repo()

in sync/worktree.py [0:0]

• 56 lines of code
• 15 McCabe index (conditional complexity)

def cleanup_repo(repo: Repo, pygit2_repo: Repository, max_count: Optional[int] = None) -> None:
    # TODO: Always cleanup repos where the sync is finished
    prune_worktrees(pygit2_repo)
    unprunable = []
    maybe_prunable = []
    prunable = []
    now = datetime.now()
    for worktree in worktrees(pygit2_repo):
        if not os.path.exists(worktree.path):
            worktree.prune(True)
            continue

        process_name = ProcessName.from_tuple(worktree.name.split("-"))

        worktree_data = (datetime.fromtimestamp(os.stat(worktree.path).st_mtime),
                         process_name,
                         worktree)

        if process_name is None:
            logger.warning("Worktree doesn't correspond to a sync %s" % worktree.path)
            unprunable.append(worktree_data)
            continue

        pygit2_repo = pygit2.Repository(worktree.path)
        head_branch = pygit2_repo.head.name

        if not head_branch or head_branch == "HEAD":
            logger.warning("No branch associated with worktree %s" % worktree.path)
            maybe_prunable.append(worktree_data)
            continue

        if head_branch.startswith("refs/heads/"):
            head_branch = head_branch[len("refs/heads/"):]

        branch_process_name = ProcessName.from_path(head_branch)
        if branch_process_name is None:
            logger.warning("No sync head associated with worktree %s" % worktree.path)
            maybe_prunable.append(worktree_data)
            continue

        if branch_process_name != process_name:
            logger.warning("Head branch doesn't match worktree %s" % worktree.path)
            maybe_prunable.append(worktree_data)
            continue

        prunable.append(worktree_data)

    if max_count and len(unprunable) > max_count:
        logger.error("Unable to cleanup worktrees, because there are too many unprunable worktrees")

    if not max_count:
        delete_count = 0
    else:
        delete_count = max(len(unprunable) + len(prunable) + len(maybe_prunable) - max_count, 0)

    prunable.sort()
    maybe_prunable.sort()
    for time, process_name, worktree in (prunable + maybe_prunable):
        assert process_name is not None
        if time < (now - timedelta(days=2)):
            logger.info("Removing worktree without recent activity %s" % worktree.path)
            delete_worktree(repo, process_name, worktree)
            delete_count -= 1
        elif delete_count > 0:
            logger.info("Removing LRU worktree %s" % worktree.path)
            delete_worktree(repo, process_name, worktree)
            delete_count -= 1
        else:
            break