[TransWarp] Component hierarchy traversal

Radek Kanovsky rk at dat.cz
Fri Oct 10 11:24:19 EDT 2003

On Wed, Oct 01, 2003 at 02:46:29PM -0400, Phillip J. Eby wrote:

> This issue in general is rather interesting.  None of the approaches to 
> dealing with it are perfect: it seems there is a distinct tension between 
> broad resource discoverability, and lazy creation of those same resources.

I think that genereal solution would require some changes in peak.binding.

Bellow is my solution of problem with special command invocation
without hierarchy traversal. Command here is specialy marked component
method that can be searched and invoked. It relies on explicit command
registration. Every command must have following signature:

    someCommand(self, [arg_with_default1 [,arg_with_default2 [...]]])

Application must have component hierarchy root that implements
ICommandRoot interface. Every command registeres itself via
registerCommand binding. CommandRoot has Main method that parses
signatures of commands and creates automaticaly appropriate help message
and allows command invocation by name and optionaly by component
path. Without argument Main outputs commands summary. Components with
registered commands must be binded with uponAssembly flag set to True.
Simple example is at the end.


Radek K

from peak.api import binding, adapt, protocols, storage, config
from peak.config.config_components import ConfigurationRoot
import sys

__all__ = ["ICommandRoot", "CommandRoot", "registerCommand"]

class ICommandRoot (config.IConfigurationRoot) :
    """Root component that is able to invoke registered commands."""

    def _registerCommand (comp, cmdname, cmd) :
        """Registers command cmd of component comp upon assembly."""
    def Main (argv) :
        """Main method."""

class registerCommand (binding.Attribute) :
    cmdname = cmd = None

    def __init__ (self, **kw) :
        if len(kw) != 1 :
            raise ValueError("One keyword argument required (name=Command)")
        self.cmdname, self.cmd = kw.items()[0]
        super(registerCommand, self).__init__(uponAssembly=True)
    def computeValue (self, comp, instanceDict, attrName) :
        root = adapt(binding.getRootComponent(comp), ICommandRoot)
        root._registerCommand(comp, self.cmdname, self.cmd)
        return self.cmd

class CommandRoot (ConfigurationRoot) :

        instancesProvide = [ICommandRoot],

    def __init__ (self, **kw) :
        super(CommandRoot, self).__init__(None, **kw)

    def uponAssembly(self, *arg, **kw) :
        # XXX Ensure that .ini files are parsed first
        super(CommandRoot, self).uponAssembly(*arg, **kw)

    _commandRegistry = binding.Make(dict)

    _options = binding.Make(dict)

    _internal_options = {"abort":1, "help":1}

    def _registerCommand (self, comp, cmdname, cmd) :
        fc = cmd.func_code
        opts = fc.co_varnames[1:fc.co_argcount]
        if cmd.func_defaults and len(opts) != len(cmd.func_defaults) :
            raise SyntaxError(
                "Command %s of component %s has positional " \
                "argument without default value" % (cmd, comp))
        for o in opts :
            if o in self._internal_options :
                raise SyntaxError(
                    "Argument %s is for internal purpose (%s of %s)" % (
                    o, cmd, comp))
            self._options[o] = 1
        cr = self._commandRegistry.setdefault(cmdname, {})
        cr[str(binding.getComponentPath(comp))] = (comp, cmd)

    def Usage (self) :
            "%s [OPTIONS] [COMMAND [COMP1 [COMP2 [...]]]]\n"
            "Avalilable commands:\n" % self._argv[0]
        cmds = self._commandRegistry.keys()
        for c in cmds :
            sys.stderr.write("  %s\n" % c)
            "Internal options:\n"
            "  --abort    call abortTransaction finally\n"
            "  --help     prints this help\n"
            "Command options:\n"
        for o in self._options.iterkeys() :
            sys.stderr.write("  --%s=PYTHON_EXPR\n" % o)

    _argv = binding.Make(lambda s,d,a: sys.argv)

    def Main (self) :
        import getopt

        try :
            opts, args = getopt.getopt(self._argv[1:], "",
                                self._internal_options.keys() +
                                ["%s=" % o for o in self._options.iterkeys()]
        except getopt.error, e :
            sys.stderr.write(str(e) + "\n")
            sys.stderr.write("try ``%s --help''\n" % self._argv[0])

        abort = False
        kw = {}
        for o,v in opts :
            if o == "--help" :
            elif o == "--abort" :
                abort = True
            else :
                kw[o[2:]] = eval(v)

        if not args :
            print "Registered commands:"
            for cmd, cr in self._commandRegistry.iteritems() :
                print "Command %r:" % cmd
                for p, comp in cr.iteritems() :
                    print "  ", p, "  ", comp[0]
        command = args[0]
        paths = args[1:]
        if not command in self._commandRegistry :
            sys.stderr.write("Command %s not registered\n" % command)
            sys.stderr.write("try ``%s --help''\n" % self._argv[0])

        if kw.get("verbose") :
            print "+"
            print "+ START TRANSACTION"
        for path, comp_cmd in self._commandRegistry[command].iteritems() :
            if paths and path not in paths :
            comp, cmd = comp_cmd
            fc = cmd.func_code
            opts = fc.co_varnames[1:fc.co_argcount]
            fkw = dict([(k,v) for k,v in kw.iteritems() if k in opts ])
            if kw.get("verbose") :
                print "+"
                print "+ COMMAND %s/%s(%s)" % (
                    path, command, 
                    ",".join([("%s=%s" % (k,v)) for k,v in fkw.iteritems()]))
            cmd(comp, **fkw)

        if abort :
            if kw.get("verbose") :
                print "+"
                print "+ ABORT TRANSACTION"
        else :
            if kw.get("verbose") :
                print "+"
                print "+ COMMIT TRANSACTION"

if __name__ == '__main__' :
    class Foo (binding.Component) :
        def cmdFoo (self, arg1=12) :
            print '%r.cmdFoo(arg1=%r)' % (self, arg1)

        cmdFoo = registerCommand(foo=cmdFoo)

    class Bar (Foo) :
        def cmdBar (self, arg1=12, arg2=42) :
            print '%r.cmdBar(arg1=%r, arg2=%r)' % (self, arg1, arg2)

        cmdBar = registerCommand(bar=cmdBar)

    class Application (CommandRoot) :

        foo = binding.Make(Foo, uponAssembly=True)
        bar = binding.Make(Bar, uponAssembly=True)


More information about the PEAK mailing list