paulin/bonobo
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Working toward sqlalchemy extension. Better ability to extend context. Still needs a lot of cleanup.

Browse Source
This commit is contained in:
Romain Dorgueil
2017-01-03 22:06:09 +01:00
parent 6bfe751c73
commit f736e7e7e3
27 changed files with 387 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
bonobo/__init__.py
View File

@ -43,11 +43,12 @@ __all__ = [
'ThreadPoolExecutorStrategy',
'__version__',
'console_run',
'head',
'inject',
'jupyter_run',
'limit',
'log',
'noop',
'pprint',
'run',
'service',
'tee',

3
bonobo/core/__init__.py
View File

@ -1,6 +1,6 @@
""" Core required libraries. """
from .bags import Bag
from .bags import Bag, ErrorBag
from .graphs import Graph
from .services import inject, service
from .strategies.executor import ThreadPoolExecutorStrategy, ProcessPoolExecutorStrategy
@ -8,6 +8,7 @@ from .strategies.naive import NaiveStrategy
__all__ = [
'Bag',
'ErrorBag',
'Graph',
'NaiveStrategy',
'ProcessPoolExecutorStrategy',

41
bonobo/core/bags.py
View File

@ -2,6 +2,11 @@ import itertools
from bonobo.util.tokens import Token
__all__ = [
'Bag',
'ErrorBag',
]
INHERIT_INPUT = Token('InheritInput')
@ -18,7 +23,8 @@ class Bag:
return self._args
return (
*self._parent.args,
*self._args, )
*self._args,
)
@property
def kwargs(self):
@ -34,7 +40,24 @@ class Bag:
return self._flags
def apply(self, func_or_iter, *args, **kwargs):
return func_or_iter(*args, *self.args, **kwargs, **self.kwargs)
if callable(func_or_iter):
return func_or_iter(*args, *self.args, **kwargs, **self.kwargs)
if len(args) == 0 and len(kwargs) == 0:
try:
iter(func_or_iter)
def generator():
nonlocal func_or_iter
for x in func_or_iter:
yield x
return generator()
except TypeError as exc:
print('nop')
raise TypeError('Could not apply bag to {}.'.format(func_or_iter)) from exc
raise TypeError('Could not apply bag to {}.'.format(func_or_iter))
def extend(self, *args, **kwargs):
return type(self)(*args, _parent=self, **kwargs)
@ -48,7 +71,13 @@ class Bag:
def __repr__(self):
return '<{} ({})>'.format(
type(self).__name__, ', '.join(
itertools.chain(
map(repr, self.args),
('{}={}'.format(k, repr(v)) for k, v in self.kwargs.items()), )))
type(self).__name__, ', '.
join(itertools.chain(
map(repr, self.args),
('{}={}'.format(k, repr(v)) for k, v in self.kwargs.items()),
))
)
class ErrorBag(Bag):
pass

112
bonobo/core/contexts.py
View File

@ -3,7 +3,7 @@ from functools import partial
from queue import Empty
from time import sleep
from bonobo.core.bags import Bag, INHERIT_INPUT
from bonobo.core.bags import Bag, INHERIT_INPUT, ErrorBag
from bonobo.core.errors import InactiveReadableError
from bonobo.core.inputs import Input
from bonobo.core.stats import WithStatistics
@ -11,11 +11,31 @@ from bonobo.util.lifecycle import get_initializer, get_finalizer
from bonobo.util.tokens import BEGIN, END, NEW, RUNNING, TERMINATED, NOT_MODIFIED
def get_name(mixed):
try:
return mixed.__name__
except AttributeError:
return type(mixed).__name__
def create_component_context(component, parent):
try:
CustomComponentContext = component.Context
except AttributeError:
return ComponentExecutionContext(component, parent=parent)
if ComponentExecutionContext in CustomComponentContext.__mro__:
bases = (CustomComponentContext, )
else:
bases = (CustomComponentContext, ComponentExecutionContext)
return type(get_name(component).title() + 'ExecutionContext', bases, {})(component, parent=parent)
class ExecutionContext:
def __init__(self, graph, plugins=None):
self.graph = graph
self.components = [ComponentExecutionContext(component, self) for component in self.graph.components]
self.components = [create_component_context(component, parent=self) for component in self.graph.components]
self.plugins = [PluginExecutionContext(plugin, parent=self) for plugin in plugins or ()]
for i, component_context in enumerate(self):
@ -23,8 +43,10 @@ class ExecutionContext:
component_context.outputs = [self[j].input for j in self.graph.outputs_of(i)]
except KeyError:
continue
component_context.input.on_begin = partial(component_context.send, BEGIN, _control=True)
component_context.input.on_end = partial(component_context.send, END, _control=True)
component_context.input.on_finalize = partial(component_context.finalize)
def __getitem__(self, item):
return self.components[item]
@ -63,17 +85,19 @@ class AbstractLoopContext:
def initialize(self):
# pylint: disable=broad-except
try:
get_initializer(self.wrapped)(self)
initializer = get_initializer(self.wrapped)
except Exception as exc:
self.handle_error(exc, traceback.format_exc())
else:
return initializer(self)
def loop(self):
"""Generic loop. A bit boring. """
while self.alive:
self._loop()
self.step()
sleep(self.PERIOD)
def _loop(self):
def step(self):
"""
TODO xxx this is a step, not a loop
"""
@ -83,9 +107,11 @@ class AbstractLoopContext:
"""Generic finalizer. """
# pylint: disable=broad-except
try:
get_finalizer(self.wrapped)(self)
finalizer = get_finalizer(self.wrapped)
except Exception as exc:
self.handle_error(exc, traceback.format_exc())
return self.handle_error(exc, traceback.format_exc())
else:
return finalizer(self)
def handle_error(self, exc, trace):
"""
@ -112,7 +138,7 @@ class PluginExecutionContext(AbstractLoopContext):
def shutdown(self):
self.alive = False
def _loop(self):
def step(self):
try:
self.wrapped.run(self)
except Exception as exc: # pylint: disable=broad-except
@ -157,13 +183,17 @@ class ComponentExecutionContext(WithStatistics, AbstractLoopContext):
return (
(
'in',
self.stats['in'], ),
self.stats['in'],
),
(
'out',
self.stats['out'], ),
self.stats['out'],
),
(
'err',
self.stats['err'], ), )
self.stats['err'],
),
)
def recv(self, *messages):
"""
@ -195,7 +225,7 @@ class ComponentExecutionContext(WithStatistics, AbstractLoopContext):
self.stats['in'] += 1
return row
def _call(self, bag):
def apply_on(self, bag):
# todo add timer
if getattr(self.component, '_with_context', False):
return bag.apply(self.component, self)
@ -203,10 +233,29 @@ class ComponentExecutionContext(WithStatistics, AbstractLoopContext):
def initialize(self):
# pylint: disable=broad-except
assert self.state is NEW, ('A {} can only be run once, and thus is expected to be in {} state at '
'initialization time.').format(type(self).__name__, NEW)
assert self.state is NEW, (
'A {} can only be run once, and thus is expected to be in {} state at '
'initialization time.'
).format(type(self).__name__, NEW)
self.state = RUNNING
super().initialize()
try:
initializer_outputs = super().initialize()
self.handle(None, initializer_outputs)
except Exception as exc:
self.handle_error(exc, traceback.format_exc())
def finalize(self):
# pylint: disable=broad-except
assert self.state is RUNNING, ('A {} must be in {} state at finalization time.'
).format(type(self).__name__, RUNNING)
self.state = TERMINATED
try:
finalizer_outputs = super().finalize()
self.handle(None, finalizer_outputs)
except Exception as exc:
self.handle_error(exc, traceback.format_exc())
def loop(self):
while True:
@ -228,34 +277,39 @@ class ComponentExecutionContext(WithStatistics, AbstractLoopContext):
output channel."""
input_bag = self.get()
outputs = self._call(input_bag)
outputs = self.apply_on(input_bag)
self.handle(input_bag, outputs)
def run(self):
self.initialize()
self.loop()
def handle(self, input_bag, outputs):
# self._exec_time += timer.duration
# Put data onto output channels
try:
outputs = _iter(outputs)
except TypeError:
except TypeError: # not an iterator
if outputs:
self.send(_resolve(input_bag, outputs))
if isinstance(outputs, ErrorBag):
outputs.apply(self.handle_error)
else:
self.send(_resolve(input_bag, outputs))
else:
# case with no result, an execution went through anyway, use for stats.
# self._exec_count += 1
pass
else:
while True:
while True: # iterator
try:
output = next(outputs)
except StopIteration:
break
self.send(_resolve(input_bag, output))
def finalize(self):
# pylint: disable=broad-except
assert self.state is RUNNING, ('A {} must be in {} state at finalization time.').format(
type(self).__name__, RUNNING)
self.state = TERMINATED
super().finalize()
else:
if isinstance(output, ErrorBag):
output.apply(self.handle_error)
else:
self.send(_resolve(input_bag, output))
def _iter(mixed):

18
bonobo/core/errors.py
View File

@ -20,9 +20,12 @@ class AbstractError(NotImplementedError):
"""Abstract error is a convenient error to declare a method as "being left as an exercise for the reader"."""
def __init__(self, method):
super().__init__('Call to abstract method {class_name}.{method_name}(...): missing implementation.'.format(
class_name=method.__self__.__name__,
method_name=method.__name__, ))
super().__init__(
'Call to abstract method {class_name}.{method_name}(...): missing implementation.'.format(
class_name=method.__self__.__name__,
method_name=method.__name__,
)
)
class InactiveIOError(IOError):
@ -39,9 +42,12 @@ class InactiveWritableError(InactiveIOError):
class ValidationError(RuntimeError):
def __init__(self, inst, message):
super(ValidationError, self).__init__('Validation error in {class_name}: {message}'.format(
class_name=type(inst).__name__,
message=message, ))
super(ValidationError, self).__init__(
'Validation error in {class_name}: {message}'.format(
class_name=type(inst).__name__,
message=message,
)
)
class ProhibitedOperationError(RuntimeError):

13
bonobo/core/inputs.py
View File

@ -48,12 +48,17 @@ class Input(Queue, Readable, Writable):
self._runlevel = 0
self._writable_runlevel = 0
self.on_initialize = noop
self.on_begin = noop
self.on_end = noop
self.on_finalize = noop
def put(self, data, block=True, timeout=None):
# Begin token is a metadata to raise the input runlevel.
if data == BEGIN:
if not self._runlevel:
self.on_initialize()
self._runlevel += 1
self._writable_runlevel += 1
@ -78,14 +83,18 @@ class Input(Queue, Readable, Writable):
data = Queue.get(self, block, timeout)
if data == END:
if self._runlevel == 1:
self.on_finalize()
self._runlevel -= 1
# callback
self.on_end()
if not self.alive:
raise InactiveReadableError('Cannot get() on an inactive {} (runlevel just reached 0).'.format(
Readable.__name__))
raise InactiveReadableError(
'Cannot get() on an inactive {} (runlevel just reached 0).'.format(Readable.__name__)
)
return self.get(block, timeout)
return data

8
bonobo/core/services.py
View File

@ -45,9 +45,11 @@ def inject(*iargs, **ikwargs):
def wrapper(target):
@functools.wraps(target)
def wrapped(*args, **kwargs):
return target(*itertools.chain(map(resolve, iargs), args),
**{ ** kwargs, ** {k: resolve(v)
for k, v in ikwargs.items()}})
return target(
*itertools.chain(map(resolve, iargs), args),
**{ ** kwargs, ** {k: resolve(v)
for k, v in ikwargs.items()}}
)
return wrapped

30
bonobo/core/strategies/executor.py
View File

@ -2,10 +2,10 @@ import time
from concurrent.futures import Executor
from concurrent.futures import ProcessPoolExecutor
from concurrent.futures import ThreadPoolExecutor
from threading import Thread
from bonobo.core.strategies.base import Strategy
from bonobo.util.tokens import BEGIN, END
from ..bags import Bag
@ -48,3 +48,31 @@ class ThreadPoolExecutorStrategy(ExecutorStrategy):
class ProcessPoolExecutorStrategy(ExecutorStrategy):
executor_factory = ProcessPoolExecutor
class ThreadCollectionStrategy(Strategy):
def execute(self, graph, *args, plugins=None, **kwargs):
context = self.create_context(graph, plugins=plugins)
context.recv(BEGIN, Bag(), END)
threads = []
# for plugin_context in context.plugins:
# threads.append(executor.submit(plugin_context.run))
for component_context in context.components:
thread = Thread(target=component_context.run)
threads.append(thread)
thread.start()
# XXX TODO PLUGINS
while context.alive and len(threads):
time.sleep(0.1)
threads = list(filter(lambda thread: thread.is_alive, threads))
# for plugin_context in context.plugins:
# plugin_context.shutdown()
# executor.shutdown()
return context

53
bonobo/ext/console/plugin.py
View File

@ -53,9 +53,10 @@ class ConsoleOutputPlugin(Plugin):
def _write(self, context, rewind):
profile, debug = False, False
if profile:
append = (('Memory', '{0:.2f} Mb'.format(memory_usage())),
# ('Total time', '{0} s'.format(execution_time(harness))),
)
append = (
('Memory', '{0:.2f} Mb'.format(memory_usage())),
# ('Total time', '{0} s'.format(execution_time(harness))),
)
else:
append = ()
self.write(context, prefix=self.prefix, append=append, debug=debug, profile=profile, rewind=rewind)
@ -77,25 +78,35 @@ class ConsoleOutputPlugin(Plugin):
for i, component in enumerate(context):
if component.alive:
_line = ''.join((
t.black('({})'.format(i + 1)),
' ',
t.bold(t.white('+')),
' ',
component.name,
' ',
component.get_stats_as_string(
debug=debug, profile=profile),
' ', ))
_line = ''.join(
(
t.black('({})'.format(i + 1)),
' ',
t.bold(t.white('+')),
' ',
component.name,
' ',
component.get_stats_as_string(
debug=debug, profile=profile
),
' ',
)
)
else:
_line = t.black(''.join((
'({})'.format(i + 1),
' - ',
component.name,
' ',
component.get_stats_as_string(
debug=debug, profile=profile),
' ', )))
_line = t.black(
''.join(
(
'({})'.format(i + 1),
' - ',
component.name,
' ',
component.get_stats_as_string(
debug=debug, profile=profile
),
' ',
)
)
)
print(prefix + _line + t.clear_eol)
if append:

8
bonobo/ext/jupyter/plugin.py
View File

@ -6,9 +6,11 @@ try:
except ImportError as e:
import logging
logging.exception('You must install Jupyter to use the bonobo Jupyter extension. Easiest way is to install the '
'optional "jupyter" dependencies with «pip install bonobo[jupyter]», but you can also install a '
'specific version by yourself.')
logging.exception(
'You must install Jupyter to use the bonobo Jupyter extension. Easiest way is to install the '
'optional "jupyter" dependencies with «pip install bonobo[jupyter]», but you can also install a '
'specific version by yourself.'
)
class JupyterOutputPlugin(Plugin):

19
bonobo/ext/opendatasoft.py
View File

@ -3,17 +3,20 @@ from urllib.parse import urlencode
import requests # todo: make this a service so we can substitute it ?
def from_opendatasoft_api(dataset=None,
endpoint='{scheme}://{netloc}{path}',
scheme='https',
netloc='data.opendatasoft.com',
path='/api/records/1.0/search/',
rows=100,
**kwargs):
def from_opendatasoft_api(
dataset=None,
endpoint='{scheme}://{netloc}{path}',
scheme='https',
netloc='data.opendatasoft.com',
path='/api/records/1.0/search/',
rows=100,
**kwargs
):
path = path if path.startswith('/') else '/' + path
params = (
('dataset', dataset),
('rows', rows), ) + tuple(sorted(kwargs.items()))
('rows', rows),
) + tuple(sorted(kwargs.items()))
base_url = endpoint.format(scheme=scheme, netloc=netloc, path=path) + '?' + urlencode(params)
def _extract_ods():

3
bonobo/io/csv.py
View File

@ -59,7 +59,8 @@ class CsvReader(CsvHandler, FileReader):
if len(row) != field_count:
raise ValueError('Got a line with %d fields, expecting %d.' % (
len(row),
field_count, ))
field_count,
))
yield dict(zip(headers, row))

5
bonobo/io/file.py
View File

@ -36,6 +36,7 @@ class FileHandler:
:param ctx:
:return:
"""
assert not hasattr(ctx, 'file'), 'A file pointer is already in the context... I do not know what to say...'
ctx.file = self.open()
@ -95,8 +96,8 @@ class FileWriter(Writer):
mode = 'w+'
def initialize(self, ctx):
super().initialize(ctx)
ctx.line = 0
return super().initialize(ctx)
def write(self, ctx, row):
"""
@ -114,4 +115,4 @@ class FileWriter(Writer):
def finalize(self, ctx):
del ctx.line
super().finalize(ctx)
return super().finalize(ctx)

93
bonobo/util/__init__.py
View File

@ -1,47 +1,116 @@
""" Various simple utilities. """
import functools
import pprint
from pprint import pprint as _pprint
import blessings
from .tokens import NOT_MODIFIED
from .helpers import run, console_run, jupyter_run
from .tokens import NOT_MODIFIED
__all__ = [
'NOT_MODIFIED',
'console_run',
'head',
'jupyter_run',
'limit',
'log',
'noop',
'pprint',
'run',
'tee',
]
def head(n=10):
def identity(x):
return x
def limit(n=10):
i = 0
def _head(x):
def _limit(*args, **kwargs):
nonlocal i, n
i += 1
if i <= n:
yield x
yield NOT_MODIFIED
_head.__name__ = 'head({})'.format(n)
return _head
_limit.__name__ = 'limit({})'.format(n)
return _limit
def tee(f):
@functools.wraps(f)
def wrapped(x):
def wrapped(*args, **kwargs):
nonlocal f
f(x)
return x
f(*args, **kwargs)
return NOT_MODIFIED
return wrapped
log = tee(pprint.pprint)
log = tee(_pprint)
def pprint(title_keys=('title', 'name', 'id'), print_values=True, sort=True):
term = blessings.Terminal()
def _pprint(*args, **kwargs):
nonlocal title_keys, term, sort, print_values
row = args[0]
for key in title_keys:
if key in row:
print(term.bold(row.get(key)))
break
if print_values:
for k in sorted(row) if sort else row:
print(
' • {t.blue}{k}{t.normal} : {t.black}({tp}){t.normal} {v}{t.clear_eol}'.format(
k=k, v=repr(row[k]), t=term, tp=type(row[k]).__name__
)
)
yield NOT_MODIFIED
_pprint.__name__ = 'pprint'
return _pprint
'''
def writehr(self, label=None):
width = t.width or 80
if label:
label = str(label)
sys.stderr.write(t.black('·' * 4) + shade('{') + label + shade('}') + t.black('·' * (width - (6+len(label)) - 1)) + '\n')
else:
sys.stderr.write(t.black('·' * (width-1) + '\n'))
def writeln(self, s):
"""Output method."""
sys.stderr.write(self.format(s) + '\n')
def initialize(self):
self.lineno = 0
def transform(self, hash, channel=STDIN):
"""Actual transformation."""
self.lineno += 1
if not self.condition or self.condition(hash):
hash = hash.copy()
hash = hash if not isinstance(self.field_filter, collections.Callable) else hash.restrict(self.field_filter)
if self.clean:
hash = hash.restrict(lambda k: len(k) and k[0] != '_')
self.writehr(self.lineno)
self.writeln(hash)
self.writehr()
sys.stderr.write('\n')
yield hash
'''
def noop(*args, **kwargs): # pylint: disable=unused-argument

12
bonobo/util/helpers.py
View File

@ -1,20 +1,20 @@
def run(*chain, plugins=None):
def run(*chain, plugins=None, strategy=None):
from bonobo import Graph, ThreadPoolExecutorStrategy
graph = Graph()
graph.add_chain(*chain)
executor = ThreadPoolExecutorStrategy()
executor = (strategy or ThreadPoolExecutorStrategy)()
return executor.execute(graph, plugins=plugins or [])
def console_run(*chain, output=True, plugins=None):
def console_run(*chain, output=True, plugins=None, strategy=None):
from bonobo.ext.console import ConsoleOutputPlugin
return run(*chain, plugins=(plugins or []) + [ConsoleOutputPlugin()] if output else [])
return run(*chain, plugins=(plugins or []) + [ConsoleOutputPlugin()] if output else [], strategy=strategy)
def jupyter_run(*chain, plugins=None):
def jupyter_run(*chain, plugins=None, strategy=None):
from bonobo.ext.jupyter import JupyterOutputPlugin
return run(*chain, plugins=(plugins or []) + [JupyterOutputPlugin()])
return run(*chain, plugins=(plugins or []) + [JupyterOutputPlugin()], strategy=strategy)