Skip to content

Instantly share code, notes, and snippets.

What would you like to do?
Processing engine
#!/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()
container = make_container(steps, {'args': args})
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))
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
return result
return get
def foo(args):
print('in foo')
return args.n
def bar(foo):
print('in bar')
return foo * 2
def baz(container):
print('in baz start')
b = container('bar')
print('in baz end', b)
if __name__ == '__main__':
$ ./
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
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment