clean-repo.py

#!/usr/bin/env python
#
#    This program is free software: you can redistribute it and/or modify
#    it under the terms of the GNU General Public License as published by
#    the Free Software Foundation, either version 3 of the License, or
#    (at your option) any later version.
#
#    This program is distributed in the hope that it will be useful,
#    but WITHOUT ANY WARRANTY; without even the implied warranty of
#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#    GNU General Public License for more details.
#
#    <http://www.gnu.org/licenses/>.
#
# F-Gaudet 2019
#
# Inspired from this post : https://github.com/aptly-dev/aptly/issues/291

from __future__ import print_function

import argparse
import re
import sys

from apt_pkg import version_compare, init_system
from subprocess import check_output, CalledProcessError

'''
Examples :
Single package :
    clean-repo --repo buster-dev --package-query cozy-stack --dry-run

All packages :
    clean-repo --repo buster-dev --package-query 'Name' --dry-run

All vault* packages :
    clean-repo --repo buster-dev --package-query 'Name (% vault-*)' --dry-run
'''


class PurgeOldVersions:
    def __init__(self):
        self.args = self.parse_arguments()

        if self.args.dry_run:
            print("Run in dry mode, without actually deleting the packages.")

        if not self.args.repo:
            sys.exit("Repo name missing.")

        if not self.args.package_query:
            sys.exit("Package name missing.")

        if self.args.keep <= 0:
            sys.exit("Please keep at least one version.")

        print("Remove \"" + self.args.package_query + "\" from " + self.args.repo +
              " and keep the last " + str(self.args.keep) +
              " packages.")


    @staticmethod
    def parse_arguments():
        parser = argparse.ArgumentParser(
            formatter_class=argparse.RawTextHelpFormatter)
        parser.add_argument("--dry-run", dest="dry_run",
                            help="List packages to remove without removing "
                                 "them.", action="store_true")
        parser.add_argument("--repo", dest="repo",
                            help="Which repository should be searched ?",
                            type=str)
        parser.add_argument("--package-query", dest="package_query",
                            help="Which package should be removed ?\n"
                                 "e.g.\n"
                                 "  - Single package: cozy-stack.\n"
                                 "  - Query: 'Name (%% vault-*)' "
                                 "to match all vault packages.\n"
                                 "  - Query: 'Name' "
                                 "to match all packages. See \n"
                                 "https://www.aptly.info/doc/feature/query/",
                            type=str)
        parser.add_argument("-k", "--keep", dest="keep",
                            help="How many package versions should be kept?",
                            type=int, default=20)
        return parser.parse_args()


    def get_packages(self):
        '''Get the list of packages for a given repository. Use the package
        query and the repository name given on the command line.

        Returns:
            Set of unique package name
        '''
        packages=set([])
        try:
            output = check_output(["aptly", "repo", "search",
                                   self.args.repo, self.args.package_query])
        except CalledProcessError as e:
            print(e)
            sys.exit()

        output = [line for line in output.split("\n") if line]
        packages.update([line.split('_')[0] for line in output])
        return packages


    def get_package(self,package_name):
        '''Get the list of package occurance. Use the repository name given
        on the command line.

        Args:
            package_name (str) : The package name

        Returns:
            List of package name with version
        '''
        try:
            output = check_output(["aptly", "repo", "search",
                                   self.args.repo, package_name])
        except CalledProcessError as e:
            print(e)
            sys.exit()
        output = [line for line in output.split("\n") if
                  line.startswith(package_name)]
        return output


    def purge_packages(self):
        '''Purge the packages depending on the repository and the
        package query given on the command line.

        '''
        init_system()
        packages = self.get_packages()
        should_keep = []
        should_delete = []

        for package_name in packages:
            package_with_all_version=self.get_package(package_name)

            def sort_cmp(name1, name2):
                version_and_build_1 = name1.split("_")[1]
                version_and_build_2 = name2.split("_")[1]
                return version_compare(version_and_build_1,
                                       version_and_build_2)

            package_with_all_version.sort(cmp=sort_cmp)
            should_delete += package_with_all_version[:-self.args.keep]
            should_keep += package_with_all_version[-self.args.keep:]

        if self.args.dry_run:
            if should_keep:
                print("\nThis package(s) would be kept:")
                for p in should_keep:
                    print(p)
            if should_delete:
                print("\nThis package(s) would be deleted:")
                for p in should_delete:
                    print(p)
        else:
            if should_delete:
                try:
                    print(check_output(["aptly", "repo", "remove",
                                    self.args.repo,
                                    " ".join(should_delete)]))
                    print(check_output(["aptly", "publish", "update",
                                    self.args.repo.split('-')[0] ] ))
                except CalledProcessError as e:
                    print(e)
                    sys.exit()
            else:
                print("Nothing to remove")


if __name__ == '__main__':
    purge_old_versions = PurgeOldVersions()
    purge_old_versions.purge_packages()

Hello,

I was searching for a tool like that .. but maybe the Aptly Syntax changed a bit (the output) or it has problems with our filenames:

ERROR: unable to remove: parsing failed: unexpected token "tool-backoffice-ui-sandbox_0.0.21+0~20201030112535_amd64": expecting end of query
ERROR: unable to remove: parsing failed: unexpected token "tool-backoffice-ui-sandbox_0.0.21+0~20201030112535_amd64": expecting end of query
Command '['aptly', 'repo', 'remove', 'stretch-testing', 'tool-backoffice-ui-sandbox_0.0.21+0~20201030101836_amd64 
....

Do you have an idea, what the root cause is ?

The -dry-run seems working fine.

cu denny