Last Updated: June 06, 2016
·
755
· themichael'tips

Python Decorator: adding events to methods

Decorator allow in a simple way the possibility to add new features at the end of methods without modifying the class's code.

Here a Workflow class with two simple methods; the rest of code show how extend the class and use decorators to adding new dynamic events at the end of the methods.

class Workflow:
    def __init__(self, name):
        self.name = name
        print "Workflow of %s" % name

    def run(self, *args, **kwargs):
        print "%s run()" % self.name

    def report(self, *args, **kwargs):
        print "%s report()" % self.name


class DecoreWorkflow(Workflow):
    tasks = {}

    def __init__(self, name):
        Workflow.__init__(self, name)

    def add_task(self, name, type):
        if self.tasks.has_key(name):
            self.tasks[name].append(type)
        else:
            self.tasks[name] = [type]

    def run(self, *args, **kwargs):
        Workflow.run(self, args, kwargs)
        [task(self) for task in self.tasks.get('run', [])]

    def report(self, *args, **kwargs):
        Workflow.report(self, args, kwargs)
        [task(self) for task in self.tasks.get('report', [])]


def runA(x):
    print "runA() after %s run()" % x.name

def runB(x):
    print "runB() after %s run()" % x.name

def reportA(x):
    print "reportA() after %s report()" % x.name

def dwork(type, name):
    def decorator(func):
        def wrappper(*args, **kwargs):
            wf = func(*args, **kwargs)
            if not isinstance(wf, Workflow):
                raise Exception('Need Workflow instance')
            wf.__class__ = DecoreWorkflow
            wf.add_task(name, type)
            return wf
        return wrappper
    return decorator

def RunDecorator(type):
    return dwork(type, "run")

def ReportDecorator(type):
    return dwork(type, "report")

@ReportDecorator(reportA)
@RunDecorator(runB)
@RunDecorator(runA)
def my_injection_function(x):
    wf = Workflow(x)
    return wf

wf = my_injection_function("<Smart Decorator>")
wf.run()
wf.report()

The output:

Workflow of <Smart Decorator>
<Smart Decorator> run()
runA() after <Smart Decorator> run()
runB() after <Smart Decorator> run()
<Smart Decorator> report()
reportA() after <Smart Decorator> report()