hsdepchk

#!/usr/bin/env python3
import argparse, functools, os, re, subprocess, sys
import requests

def get_dep_checker_url(name):
    if not re.match(r"[A-Za-z0-9._-]+", name):
        raise ValueError("invalid package name: {}".format(name))
    return "http://packdeps.haskellers.com/feed?needle=exact%3A" + name

def is_pkg_nonrestrictive(name, timeout=60.0, max_size=(64 * 1024)):
    '''Check if the package has non-restrictive version upper bounds on
    packdeps.haskells.com.'''
    r = requests.get(get_dep_checker_url(name), stream=True, timeout=timeout)
    r.raise_for_status()
    body = b""
    for chunk in r.iter_content(max_size + 1):
        body = chunk
    if len(body) > max_size:
        raise Exception("response body exceeded {} bytes".format(max_size))
    if b"No packages checked" in body:
        raise ValueError("cannot find package: {}".format(name))
    return b"All upper bounds are non-restrictive." in body

def is_is_exception_fatal(e):
    if isinstance(e, AttributeError):
        return True
    if isinstance(e, NameError):
        return True
    if isinstance(e, TypeError):
        return True
    if isinstance(e, ValueError):
        return True
    if isinstance(e, requests.HTTPError):
        return e.response.status_code // 100 == 4
    return False

def retry(attempts, action):
    if attempts <= 0:
        raise ValueError("attempts must be positive: {}".format(attempts))
    for attempt_index in range(attempts):
        try:
            return action()
        except Exception as e:
            if is_is_exception_fatal(e) or attempt_index >= attempts - 1:
                raise

class DepChecker(object):

    def __init__(self, pkg_names, attempts):
        self.pkg_names = pkg_names
        self.attempts = attempts

    def run(self):
        self.success = True
        for name in self.pkg_names:
            retry(self.attempts, functools.partial(self.check_pkg, name))
        return self.success

    def check_pkg(self, name):
        if not is_pkg_nonrestrictive(name):
            url = get_dep_checker_url(name)
            sys.stderr.write("Package '{}' contains restrictive bounds. "
                             "See: {}\n".format(name, url))
            sys.stderr.flush()
            self.success = False

def main():
    p = argparse.ArgumentParser()
    p.add_argument("-n", "--attempts", default=5)
    p.add_argument("pkg_names", metavar="pkg-name", nargs="*")
    app = DepChecker(**vars(p.parse_args()))
    return int(not app.run())

if __name__ == "__main__":
    sys.exit(main())