Help tool for managing packages contained in Conda channels

channel_helper.py

#!/usr/bin/env python3
### Usage:
###   ./get_packages.py get url arch
### Examples:
###   ./get_packages.py get https://conda.anaconda.org/idaholab linux-64
import requests, re, sys, argparse, os

from urllib3.exceptions import InsecureRequestWarning
requests.packages.urllib3.disable_warnings(category=InsecureRequestWarning)

class ChannelHelper:
    def __init__(self, url):
        self.url = url

    def get_url(self, arch):
        url = '/'.join([self.url, arch, 'repodata.json'])
        results = requests.get(url, verify=False)
        if results.status_code == 200:
            return results.json()
        # don't care about errors
        sys.exit(0)

    def rPackage(self):
        return re.compile(r'([\w+-]+)-\d.*')

    def get_packages(self, arch):
        results = self.get_url(arch)
        packages = {}
        for k, v in results['packages'].items():
            binary = self.rPackage().findall(k)[0]
            packages[binary] = packages.get(binary, { 'binaries' : [], 'timestamps' : {}})
            packages[binary]['binaries'].append(k)
            packages[binary]['timestamps'].update( { v['timestamp'] : k } )
        return packages

    def order_packages(self, arch, package=None, oldest=None, newest=None):
        """ yield packages to caller (this is a generator) """
        packages = self.get_packages(arch)

        # do something with a single package
        if package:
            try:
                query = list(packages[package]['timestamps'].keys())
            except KeyError:
                print('Package %s not available' % (package))
                sys.exit(1)

            # Sort oldest first
            query.sort()

            # yield the oldest version of this package
            if oldest:
                yield packages[package]['timestamps'][query[0]]
            # yield the newest version of this package
            elif newest:
                yield packages[package]['timestamps'][query[-1]]
            # yield every version of this package
            else:
                for timestamp in query:
                    yield packages[package]['timestamps'][timestamp]

        # do something with every package
        else:
            for package in packages.keys():
                timestamps = list(packages[package]['timestamps'].keys())
                timestamps.sort()
                # yield oldest of every package
                if oldest:
                    yield packages[package]['timestamps'][timestamps[0]]
                # yield newest of every package
                elif newest:
                    yield packages[package]['timestamps'][timestamps[-1]]
                # yield every package
                else:
                    for timestamp in timestamps:
                        yield packages[package]['timestamps'][timestamp]

def verifyArgs(parser):
    args = parser.parse_args()
    if not args.url or not args.arch:
        parser.print_help()
        sys.exit(1)
    if args.command == 'get' and (args.meta_delete and os.path.dirname(args.url) != "https://conda.anaconda.org"):
        print('-d --delete only works with Anaconda channels')
        sys.exit(1)
    if args.command == 'diff':
        if args.meta_server is None:
            print('-s|--server cannot be empty')
            sys.exit(1)
    return args

def parseArgs(args=None):
    parser = argparse.ArgumentParser(description='A tool to help manage personal Conda channels. As well as printing commands to interface with Anaconda channels')
    formatter = lambda prog: argparse.HelpFormatter(prog, max_help_position=55, width=130)
    parent = argparse.ArgumentParser(add_help=False)
    parent.add_argument('url', nargs="?", help='URL to Channel')
    parent.add_argument('arch', nargs="?", help='Package Architecture (linux-64, osx-64)')

    # Use subparser so we can support more commands one day
    subparser = parser.add_subparsers(dest='command', help='Available Commands.')
    subparser.required = True

    # get commands
    user_parser = subparser.add_parser('get', parents=[parent], help='Get Packages', formatter_class=formatter)
    user_parser.add_argument('-p', '--package', dest='meta_package', metavar='name', help='Get only this package')
    user_parser.add_argument('-o', '--oldest', dest='meta_oldest', action='store_true', help='Get the oldest package')
    user_parser.add_argument('-n', '--newest', dest='meta_newest', action='store_true', help='Get the latest package')
    user_parser.add_argument('-d', '--delete', dest='meta_delete', action='store_true', help='Print Anaconda command neccessary to delete a package')

    # transfer commands
    user_parser = subparser.add_parser('transfer', parents=[parent], help='Print transfer commands from URL to Anaconda', formatter_class=formatter)
    user_parser.add_argument('-p', '--package', dest='meta_package', metavar='name', help='Transfer only this package')
    user_parser.add_argument('-o', '--oldest', dest='meta_oldest', action='store_true', help='Get the oldest package')
    user_parser.add_argument('-n', '--newest', dest='meta_newest', action='store_true', help='Get the latest package')

    return verifyArgs(parser)

if __name__ == '__main__':
    args = parseArgs()
    cHelper = ChannelHelper(args.url)
    for package in cHelper.order_packages(args.arch,
                                              package=args.meta_package,
                                              newest=args.meta_newest,
                                              oldest=args.meta_oldest):
        if args.command == 'get':
            if args.meta_delete:
                print('anaconda remove %s/%s' % (os.path.basename(args.url), package))
            else:
                print(package)
        elif args.command == 'transfer':
            print('echo {pack}... && curl --insecure -s -L -O {}/{}/{pack} && anaconda upload --force --no-progress {pack} >/dev/null 2>&1 && rm -f {pack}'.format(args.url, args.arch, pack=package))