cfget 0.18-1 / usr / bin / cfget

#!/usr/bin/python
# -*- coding: utf-8 -*-

from optparse import OptionParser
import sys, os, os.path
import re
import imp
import pipes
import cPickle as pickle
import __builtin__

VERSION="0.18"

# Functions to report different levels of feedback. They will be redefined
# according to command line options
def verbose(*args):
    pass
def warning(*args):
    pass

#
# Exceptions that will also be exported to plugins
#

class UserError(Exception):
    """
    Exception that does not cause a backtrace, but shows as an error message to
    the user
    """
    pass

class ParseError(UserError):
    """
    Exception for parse errors, with file name and line number
    """
    def __init__(self, fname, line, msg):
        super(ParseError, self).__init__("%s:%d: %s" % (fname, line, msg))

# Make exceptions available to plugins
__builtin__.cfget = imp.new_module("cfget")
__builtin__.cfget.UserError = ParseError
__builtin__.cfget.ParseError = ParseError

#
# Built-in config file parsers
#

# ConfigParser is such an embarassing pain that it's easier to roll our own
def parse_ini_file(pathname):
    "Normal .ini file format"
    RE_EMPTY = re.compile(r"^\s*(?:[#;].*)?$")
    RE_SECTION = re.compile(r"^\s*\[([^\]]+)\]\s*$")
    RE_VALUE = re.compile(r"^\s*([^=]+)=(.*)$")

    section = None
    for lineno, line in enumerate(open(pathname)):
        if RE_EMPTY.match(line): continue
        mo = RE_SECTION.match(line)
        if mo:
            section = mo.group(1).lower().strip()
            continue
        mo = RE_VALUE.match(line)
        if mo:
            key, value = mo.group(1, 2)
            if section is None:
                path = key.strip().lower()
            else:
                path = section + "/" + key.strip().lower()
            value = value.strip()
            if value:
                yield path, value
            else:
                yield path, None
        else:
            raise ParseError(pathname, lineno+1, "line not a comment, and not in the form [section] or key=value")

# configobj based parser, with handy configobj features enabled
try:
    from configobj import ConfigObj
    def parse_configobj(pathname):
        ".ini-style file format as extended by ConfigObj (see http://www.voidspace.org.uk/python/configobj.html)"
        def yield_section(section, prefix = None):
            for k, v in section.iteritems():
                if prefix:
                    path = prefix + "/" + k
                else:
                    path = k

                if isinstance(v, dict):
                    for k, v in yield_section(v, path):
                        yield k, v
                else:
                    yield path, v
        return yield_section(ConfigObj(pathname, interpolation="template"))
except ImportError:
    parse_configobj = None

#
# Built-in dumpers
#

def dump_exports(cfget, outfd, paths):
    """
    Dump values as sh export statements
    """
    RE_NORMALISE=re.compile(r"[^A-Za-z_]+")
    for path, val in cfget.iteritems(paths):
        name = RE_NORMALISE.sub("_", path).upper()
        val = pipes.quote(str(val))
        print >>outfd, "export %s=%s" % (name, val)

def dump_repr(cfget, outfd, paths):
    """
    Dump values using python's repr() formatter
    """
    for path, val in cfget.iteritems(paths):
        print >>outfd, "%s: %s" % (path, repr(val))

def dump_pickle(cfget, outfd, paths):
    """
    Dump values as a pickled dict
    """
    res = dict(cfget.iteritems(paths))
    pickle.dump(res, outfd, -1)

#
# Built-in template engines
#

def autoconf_replacer(cfget, fd_in, fd_out):
    """
    Replace values represented as @SECTION_KEY@
    """
    RE_LINE = re.compile(r"@([^@]+)@")

    def subst(mo):
        vals = mo.group(1).split("_", 1)
        if len(vals) == 1:
            query = mo.group(1)
        else:
            query = "/".join(vals)
        return str(cfget.query(query))

    for line in fd_in:
        fd_out.write(RE_LINE.sub(subst, line))

#
# The expression interpreter
#

# Set to True to trace the interpreter
if False:
    parsefunc_nest = 0
    def parsefunc(func):
        def func1(self, tokens):
            global parsefunc_nest
            print >>sys.stderr, " "*parsefunc_nest, "TRY", func.__name__, "ON", repr(tokens)
            parsefunc_nest += 1
            num, val = func(self, tokens)
            parsefunc_nest -= 1
            if num == 0:
                print >>sys.stderr, " "*parsefunc_nest, "FAIL"
            else:
                print >>sys.stderr, " "*parsefunc_nest, "GOT", val, "NEXT", repr(tokens[num:])
            return num, val
        return func1
else:
    def parsefunc(func):
        return func

class Interpreter(object):
    """
    Interpreter to run the query expressions

    This is hand-crafted because it's simple and in order to avoid external
    dependencies.
    """
    class ParseError(Exception):
        pass

    RE_CURLS = re.compile(r"{([^{}]+)}")

    # Lexer tokens
    TOK_FUNC = 1
    TOK_LOOKUP = 2
    TOK_GROUPS = 3
    TOK_GROUPE = 4
    TOK_MATH = 5
    TOK_NUM = 6

    # Lexer
    LEX = (
        (re.compile(r"(?:^|\s+)[0-9]+(?:\.[0-9]+)?(?=\s|\)|$)"), TOK_NUM),
        (re.compile(r"\s+(?:\+|-|\*|/|\*\*)(?=\s|$)"), TOK_MATH),
        (re.compile(r"\s*[A-Za-z_]+(?=\()"), TOK_FUNC),
        (re.compile(r"\s*[^(){} ]+(?!\()"), TOK_LOOKUP),
        (re.compile(r"\s*\("), TOK_GROUPS),
        (re.compile(r"\s*\)"), TOK_GROUPE),
    )

    def __init__(self, db):
        self.db = db

    def toks2str(self, toks):
        "Format a sequence of tokens into an expression"
        res = []
        for t, v in toks:
            if t == self.TOK_MATH:
                res.append(" " + v + " ")
            else:
                res.append(v)
        return "".join(res)

    def parse_error(self, msg, pre, post):
        "Raise a parse error at the current parsing position"
        pos = self.toks2str(pre) + " (HERE) " + self.toks2str(post)
        raise self.ParseError("%s parsing expression [%s]" % (msg, pos))

    def tokenize(self, expr):
        "Split an expression in tokens"
        orig = expr
        res = []
        while expr:
            found = False
            for r, t in self.LEX:
                mo = r.match(expr)
                if mo:
                    res.append((t, mo.group(0).strip()))
                    expr = expr[mo.end(0):]
                    found = True
                    break
            if not found:
                parsepos = len(orig)-len(expr)
                raise self.ParseError("Cannot understand expression: \"%s(HERE)%s\"" % (
                    orig[:parsepos], orig[parsepos:]))
        return res

    @parsefunc
    def do_parens(self, tokens):
        "Parse '( expr )'"
        try:
            tok, val = tokens[0]
            if tok != self.TOK_GROUPS:
                return 0, None
            toks, res = self.do_expr(tokens[1:])
            if toks == 0:
                return 0, None
            tok, val = tokens[toks + 1]
            if tok != self.TOK_GROUPE:
                return 0, None
            return toks + 2, res
        except IndexError:
            return 0, None

    @parsefunc
    def do_function(self, tokens):
        "Parse 'funcname ( expr )'"
        try:
            tok, name = tokens[0]
            if tok != self.TOK_FUNC:
                return 0, None
            toks, val = self.do_parens(tokens[1:])
            if toks > 0:
                return toks+1, self.db.func(name)(self.db, val)
            return 0, None
        except IndexError:
            return 0, None

    @parsefunc
    def do_value(self, tokens):
        "Parse an atomic value"
        # A value is...
        if not tokens: return 0, None

        if tokens[0][0] == self.TOK_NUM:
            # ...a constant number
            return 1, float(tokens[0][1])
        elif tokens[0][0] == self.TOK_LOOKUP:
            # ...a section/key lookup
            return 1, self.db.lookup(tokens[0][1])
        elif tokens[0][0] == self.TOK_GROUPS:
            # ...an expression in parenthesis
            toks, val = self.do_parens(tokens)
            if toks: return toks, val
        elif tokens[0][0] == self.TOK_FUNC:
            # ...a function
            toks, val = self.do_function(tokens)
            if toks: return toks, val

        return (0, None)

    @parsefunc
    def do_expr(self, tokens):
        "Parse an expression of atomic values"
        # An expression is value [op value]*

        if not tokens: return 0, None

        # Parse everything into a (value[, op value]*) list
        expr = []

        # First value
        toks, val = self.do_value(tokens)
        if not toks: return 0, None
        expr.append(val)

        # [op, value]*
        while len(tokens) > toks and tokens[toks][0] == self.TOK_MATH:
            # op
            op, valop = tokens[toks]
            expr.append(valop)
            toks += 1

            # val
            toks2, val = self.do_value(tokens[toks:])
            if toks2 == 0:
                self.parse_error("Math operator must be followed by some value", tokens[:toks], tokens[toks:])
            try:
                expr.append(float(val))
            except (TypeError, ValueError):
                self.parse_error("Value \"%s\" did not convert to a number" % val, tokens[:toks], tokens[toks:])
            toks += toks2

        # If we have only one item, return it as it is
        if len(expr) == 1:
            return toks, expr[0]

        # If we have more, then we are a math expr and we convert also the
        # first item to a number
        try:
            expr[0] = float(expr[0])
        except (TypeError, ValueError):
            self.parse_error("Value \"%s\" did not convert to a number" % expr[0], [], tokens)

        # Build the expression and have python evaluate it. It is a nice
        # shortcut: we get evaluation order right, and it is safe to do because
        # the expression is guaranteed to be only a sequence of floats and
        # python operators
        val = eval(" ".join(map(str, expr)))
        return toks, val

    def run(self, expr):
        "Run an expression through the interperter and return its result"
        # Perform curly brace expansions
        while True:
            expr1 = self.RE_CURLS.sub(lambda mo:self.run(mo.group(1)), expr)
            if expr == expr1: break
            expr = expr1

        tokens = self.tokenize(expr)
        #print >>sys.stderr, "TOKENS", repr(tokens)
        (toks, val) = self.do_expr(tokens)
        if toks < len(tokens):
            self.parse_error("Query is badly formed", tokens[:toks], tokens[toks:])
        return val

#
# The cfget engine
#

class Cfget(object):
    class QueryStack(object):
        "Query Cfget avoiding loops"
        def __init__(self, db, seen=[]):
            self.db = db
            self.seen = set(seen)

        def query(self, path):
            return self.db.lookup(path, self)

        def __str__(self):
            return "[" + ", ".join(self.seen) + "]"

    def __init__(self, root=None):
        self.vals = dict()
        self.staticdb = dict()
        self.dynamicdb = dict()
        self.formats = dict()
        self.functions = dict()
        self.dumpers = dict()
        self.templaters = dict()
        self.root = root.strip("/") + "/" if root else None
        self.add_format("ini", parse_ini_file)
        if parse_configobj:
            self.add_format("configobj", parse_configobj)
        self.add_dumper("exports", dump_exports)
        self.add_dumper("repr", dump_repr)
        self.add_dumper("pickle", dump_pickle)
        self.add_templater("autoconf", autoconf_replacer)
        self.add_function("query", lambda db, x: db.query(x))
        self.add_function("int", lambda db, x: int(x))
        self.add_function("round", lambda db, x: int(round(x)))
        self.interpreter = Interpreter(self)

    def normalise_path(self, path):
        return path.lower()

    def load_file(self, style, pathname):
        """
        Load values from an .ini file
        """
        parser = self.formats.get(style, None)
        if parser is None:
            raise UserError("parser '%s' not found" % style)
        for path, value in parser(pathname):
            path = self.normalise_path(path)
            if path is None: continue
            if value is None:
                self.staticdb.pop(path, None)
            else:
                self.staticdb[path] = value

    def load_plugin(self, pathname):
        """
        Load a plugin or a directory of plugins
        """
        if not os.path.exists(pathname): return
        # if pathname is a dir, load all files inside
        if os.path.isdir(pathname):
            for f in os.listdir(pathname):
                if f.startswith("."): continue
                if not f.endswith(".py"): continue
                child = os.path.join(pathname, f)
                if not os.path.isfile(child): continue
                self.load_plugin(child)
        else:
            verbose("Reading plugin %s" % pathname)
            name = os.path.basename(pathname)
            name = os.path.splitext(name)[0]
            module = imp.load_source(name, pathname)
            module.UserError = UserError
            module.verbose = verbose
            module.warning = warning
            module.init(self)

    def load_from_env(self, style="ini", cfgenv="CFGET_CFG", pluginenv="CFGET_PLUGINS"):
        """
        Load config files and plugins taking their names from the environment
        """
        # Get the list of input files
        inputs = os.environ.get(cfgenv, "").split(":")
        if not inputs:
            raise UserError, "No configuration files specified in %s" % cfgenv

        # Get the list of plugins
        plugins = os.environ.get(pluginenv, "").split(":")

        # Load the plugins
        for fname in plugins:
            if not fname: continue
            self.load_plugin(fname)

        # Parse the input files
        for fname in inputs:
            if not fname: continue
            self.load_file(style, fname)

    def add_static(self, path, value):
        """
        Add a static path=value mapping, overriding existing ones
        """
        path = self.normalise_path(path)
        if path is None: return
        self.staticdb[path] = value

    def add_dynamic(self, path, func):
        """
        Add a dynamic path=function mapping, to be tried after the existing ones
        """
        path = self.normalise_path(path)
        if path is None: return
        self.dynamicdb.setdefault(path, []).append(func)

    def add_format(self, name, obj):
        """
        Add a config file parser
        """
        self.formats[name] = obj

    def add_dumper(self, path, obj):
        """
        Add a dumper to dump the contents of the database
        """
        self.dumpers[path] = obj

    def add_templater(self, path, obj):
        """
        Add a templater to perform substitutions in a template file using a
        database
        """
        self.templaters[path] = obj

    def add_function(self, name, func):
        """
        Add a function to transform a result
        """
        self.functions[name] = func

    def list_matching_re(self, expr):
        """
        List the paths that match the given regular expression
        """
        res = set()
        for k in self.staticdb.iterkeys():
            if expr.match(k):
                res.add(k)
        for k in self.dynamicdb.iterkeys():
            if expr.match(k):
                res.add(k)
        return sorted(res)

    def func(self, name):
        res = self.functions.get(name, None)
        if res is None:
            raise UserError("Function `%s' is not defined" % name)
        return res

    def query(self, expr):
        return self.interpreter.run(expr)

    def lookup(self, path, stack=None):
        """
        Query the value for a path.

        If there is a static mapping for it, return its value.

        If there are dynamic mappings for it, try them all in sequence until
        one returns a value.

        Return None if no value was found, either statically or dynamically.
        """
        path = path.lower()
        # If we have a root path, prepend it here
        if stack is None and self.root is not None:
            path = self.root + path

        # First try to query the static database
        res = self.staticdb.get(path, None)
        if res is not None: return res

        # Avoid loops
        if stack is not None:
            if path in stack.seen:
                verbose("Loop detected: %s requested itself" % path)
                return None
            stack.seen.add(path)

        # If it fails, try dynamically
        funcs = self.dynamicdb.get(path, None)
        if funcs is None: return None

        # Try all dynamic functions in turn
        for func in funcs:
            if stack is None:
                stack = self.QueryStack(self, [path])
            val = func(stack, path)
            if val is not None:
                self.staticdb[path] = val
                return val

        # All failed, return None
        return None

    def iteritems(self, paths=[]):
        """
        Iterate all path, value pairs.

        Paths will be generated sorted in alphabetical order.
        """
        if not paths:
            paths = set()
            paths.update(self.staticdb.iterkeys())
            paths.update(self.dynamicdb.iterkeys())
            if self.root is None:
                paths = list(paths)
            else:
                # If we have a root path, chop it here (query will prepend it)
                paths = [p[len(self.root):] for p in paths if p.startswith(self.root)]
        paths.sort()
        for path in paths:
            val = self.query(path)
            if val is None: continue
            yield path, val

    def dump(self, style, out=sys.stdout, paths=[]):
        "Dump all the contents to the given output stream"
        dumper = self.dumpers.get(style, None)
        if dumper is None:
            raise UserError("dump style %s not found" % style)
        dumper(self, out, paths)

    def template(self, style, fd_in, fd_out):
        tpl = self.templaters.get(style, None)
        if tpl is None:
            raise UserError("template style %s not found" % style)
        tpl(self, fd_in, fd_out)

    def dump_list(self):
        for k, v in self.dumpers.iteritems():
            print "%s: %s" % (k, v.__doc__.strip().split("\n")[0])

    def template_list(self):
        for k, v in self.templaters.iteritems():
            print "%s: %s" % (k, v.__doc__.strip().split("\n")[0])

    def format_list(self):
        for k, v in self.formats.iteritems():
            print "%s: %s" % (k, v.__doc__.strip().split("\n")[0])

#
# cfget user functions
#

def do_dump(cfget, opts, args):
    if opts.dump == "list":
        print "Available dump styles:"
        print ""
        cfget.dump_list()
    else:
        # Dump all, or the requested paths
        cfget.dump(opts.dump, paths=args)

def do_template(cfget, opts, args):
    if opts.template == "list":
        print "Available template styles:"
        print ""
        cfget.template_list()
    else:
        if len(args) > 0:
            fd_in = open(args[0])
        else:
            fd_in = sys.stdin
        if len(args) > 1:
            # Read the current umask
            umask = os.umask(0)
            os.umask(umask)
            # Create a temporary file in same directory as args[1]
            import tempfile
            outdir = os.path.dirname(os.path.abspath(args[1]))
            fd_out, tmpfname = tempfile.mkstemp(dir = outdir)
            # Chmod to something sane using the umask
            os.chmod(tmpfname, 0666 & ~umask)
            fd_out = os.fdopen(fd_out, "w")
            try:
                # Write to the temporary file
                cfget.template(opts.template, fd_in, fd_out)
                # Rename when finished
                os.rename(tmpfname, args[1])
            except:
                # Just delete the temp file if anything went wrong
                os.unlink(tmpfname)
                raise
        else:
            cfget.template(opts.template, fd_in, sys.stdout)

def do_query(cfget, opts, args):
    # Parse the requested path
    if not args:
        parser.error("Please specify a configuration value to read in the form 'section/key'")

    for path in args:
        # Query the config files in order
        res = cfget.query(path)
        if res is not None:
            print str(res)
        else:
            raise UserError("%s not found" % path)


if __name__ == "__main__":
    class Parser(OptionParser):
        "Command line parser"
        def __init__(self, *args, **kwargs):
            OptionParser.__init__(self, *args, **kwargs)

        def error(self, msg):
            sys.stderr.write("%s: error: %s\n\n" % (self.get_prog_name(), msg))
            self.print_help(sys.stderr)
            sys.exit(2)

    default_format = os.environ.get("CFGET_FORMAT", "ini")
    parser = Parser(usage="usage: %prog [options] section/key\n"
                          "   or: %prog [options] --dump=STYLE [section/key [section/key...]]\n"
                          "   or: %prog [options] --template=STYLE [infile [outfile]]",
                    version="%prog "+ VERSION,
                    description="Get values from a config file.")
    parser.add_option("-q", "--quiet", action="store_true", help=
            "quiet mode: only output fatal errors")
    parser.add_option("-v", "--verbose", action="store_true", help="verbose mode")
    parser.add_option("--debug", action="store_true", help="verbose mode")
    parser.add_option("-C","--cfg", action="append", metavar="file", help=
            "config file to read; the option can be given more than once to "
            "read more than one file. If missing, read a colon separated list "
            "from the CFGET_CFG env variable.")
    parser.add_option("-P","--plugin", action="append", metavar="file", help=
            "list of plugin files or directories to load. The option can be "
            "given more than once to read more than one file. If missing, "
            "read a colon separated list from the CFGET_PLUGINS env variable.")
    parser.add_option("-d","--dump", action="store", metavar="name", help=
            "dump the contents of the database using the given style. Use "
            "'--dump=list' for a list of available styles. If one or more "
            "paths are provided in the command line, dump only those paths, "
            "otherwise dump all.")
    parser.add_option("-t","--template", action="store", metavar="name", help=
            "read a template file, expand template placeholders using the "
            "configuration data and output the result. Use '--template=list' "
            "for a list of available styles.")
    parser.add_option("-f","--format", action="store", metavar="name",
            default=default_format, help=
            "use a custom configuration file format (default: %default). Use "
            "'--format=list' for a list of available formats. The CFGET_FORMAT "
            "environment value, if defined, can be used to provide a different "
            "default value.")
    parser.add_option('-r',"--root", action="store", metavar="path", help=
            "restrict all work to values under the given path")

    (opts, args) = parser.parse_args()

    if opts.verbose:
        def verbose(*args):
            print >>sys.stderr, " ".join(args)
    if not opts.quiet:
        def warning(*args):
            print >>sys.stderr, " ".join(args)

    # Get the list of input files
    inputs = opts.cfg
    if not inputs:
        inputs = os.environ.get("CFGET_CFG", "").split(":")
    if not inputs:
        parser.error("No configuration files specified in the command line or in CFGET_CFG")

    # Get the list of plugins
    plugins = opts.plugin
    if not plugins:
        plugins = os.environ.get("CFGET_PLUGINS", "").split(":")

    try:
        cfget = Cfget(root=opts.root)

        # Load the plugins
        for fname in plugins:
            if not fname: continue
            cfget.load_plugin(fname)

        if opts.format == "list":
            cfget.format_list()
            sys.exit(0)

        # Parse the input files
        for fname in inputs:
            if not fname: continue
            cfget.load_file(opts.format, fname)

        if opts.dump is not None:
            do_dump(cfget, opts, args)
        elif opts.template is not None:
            do_template(cfget, opts, args)
        else:
            do_query(cfget, opts, args)
    except IOError, e:
        warning("%s: %s" % (e.filename, e.strerror))
        sys.exit(1)
    except UserError, e:
        if opts.debug: raise
        warning(str(e))
        sys.exit(1)

    sys.exit(0)