Remove all refs from a local flatpak repository, except installed ones

flatpak-cleanup-local-repo.py

#!/usr/bin/env python3

import argparse
import logging
import subprocess
import typing as t

import gi
gi.require_version('GLib', '2.0')
gi.require_version('Gio', '2.0')
gi.require_version('OSTree', '1.0')
gi.require_version('Flatpak', '1.0')
from gi.repository import GLib
from gi.repository import Gio
from gi.repository import OSTree
from gi.repository import Flatpak


log = logging.getLogger(__name__)


def refs_installed_from(path: Gio.File) -> t.Iterator[str]:
    repo_uri = path.get_uri()
    installations = [Flatpak.Installation.new_user()]
    installations += Flatpak.get_system_installations()
    for flatpak_inst in installations:
        remotes = flatpak_inst.list_remotes_by_type([Flatpak.RemoteType.STATIC])
        try:
            flatpak_remote = next(r for r in remotes if r.get_url() == repo_uri)
        except StopIteration:
            continue
        for ref in flatpak_inst.list_installed_refs():
            if ref.get_origin() == flatpak_remote.get_name():
                yield ref.format_ref()


def traverse_commit_parents(repo: OSTree.Repo, commit: str) -> t.Iterator[str]:
    _success, commit_variant, _state = repo.load_commit(commit)
    assert _success
    yield commit
    parent = OSTree.commit_get_parent(commit_variant)
    while parent:
        try:
            _success, parent_variant, _state = repo.load_commit(parent)
        except GLib.Error as err:
            if err.matches(Gio.io_error_quark(), Gio.IOErrorEnum.NOT_FOUND):
                log.debug("Can't load parent commit (already deleted?): %s",
                          err.message)  # pylint: disable=no-member
                return
            raise
        assert _success
        yield parent
        parent = OSTree.commit_get_parent(parent_variant)



def main():
    parser = argparse.ArgumentParser(description="Remove everything from an OSTree repo, "
                                                 "except for flatpak refs installed from it.")
    parser.add_argument("repo", help="Local repo path, e.g. .flatpak-builder/cache")
    args = parser.parse_args()

    logging.basicConfig(level=logging.INFO)

    repo_path = Gio.File.new_for_path(args.repo)
    repo = OSTree.Repo.new(repo_path)
    repo.open()
    log.info("Opened repo at %s", repo.get_path().get_path())

    installed_refs = set(refs_installed_from(repo.get_path()))
    _success, ostree_refs = repo.list_refs()
    assert _success

    for ref, commit in ostree_refs.items():
        if ref in installed_refs:
            log.info("Found installed ref %s", ref)
            history = list(reversed(list(traverse_commit_parents(repo, commit))))
            for ansestor in history[:-1]:
                log.info("Removing commit %s", ansestor)
                repo.delete_object(OSTree.ObjectType.COMMIT, ansestor)
            continue

        log.info("Removing ref %s", ref)
        repo.set_ref_immediate(remote=None, ref=ref, checksum=None)

    log.info("Pruning repo at %s", repo.get_path().get_path())
    _success, n_total, n_pruned, s_pruned = repo.prune(flags=OSTree.RepoPruneFlags.REFS_ONLY,
                                                       depth=-1)
    assert _success
    log.info("Pruned %i/%i objects, freed %.2f MiB", n_pruned, n_total, s_pruned / 1024**2)


if __name__ == "__main__":
    main()