Processing engine

dependency_container.py

#!/usr/bin/env python3

'''
Inspired by pytest fixtures :)
'''

import argparse
from inspect import signature
import logging
from reprlib import repr as smart_repr


logger = logging.getLogger(__name__)


def main():
    p = argparse.ArgumentParser()
    p.add_argument('-n', type=int, default=3)
    args = p.parse_args()
    logging.basicConfig(level=logging.DEBUG)
    container = make_container(steps, {'args': args})
    container('baz')


steps = {}


def step(f):
    steps[f.__name__] = f
    return f


def make_container(factories, extra_values=None):
    extra_values = extra_values or {}
    cached_values = {}
    processing = set()

    def get(key, force_execute=False):
        if key in extra_values:
            return extra_values[key]
        if key == 'container':
            return get # return this function :)
        if not force_execute and key in cached_values:
            return cached_values[key]
        return execute(key)

    def execute(key):
        if key in processing:
            raise Exception('Already processing {!r} - probably circular dependency'.format(key))
        processing.add(key)
        try:
            f = factories[key]
            assert callable(f)
            sig = signature(f)
            param_names = sig.parameters.keys()
            param_data = {name: get(name) for name in param_names}
            logger.debug('Executing %s %s', f.__name__, smart_repr(param_data))
            result = f(**param_data)
            cached_values[key] = result
        finally:
            processing.remove(key)
        return result

    return get


@step
def foo(args):
    print('in foo')
    return args.n


@step
def bar(foo):
    print('in bar')
    return foo * 2


@step
def baz(container):
    print('in baz start')
    b = container('bar')
    print('in baz end', b)


if __name__ == '__main__':
    main()

output.txt

$ ./demo.py
DEBUG:__main__:Executing baz {'container': <function mak...t 0x102991e18>}
in baz start
DEBUG:__main__:Executing foo {'args': Namespace(n=3)}
in foo
DEBUG:__main__:Executing bar {'foo': 3}
in bar
in baz end 6