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

Merge branch 'develop' into dev_graphviz

Browse Source
This commit is contained in:
Romain Dorgueil
2017-08-11 07:38:53 +02:00
parent be844c3ed7 423a75d554
commit ff0d6219a6
139 changed files with 6088 additions and 1669 deletions
Download Patch File Download Diff File Expand all files Collapse all files

87
bonobo/_api.py
View File

@ -1,10 +1,10 @@
import warnings
import logging
from bonobo.basics import Limit, PrettyPrint, Tee, count, identity, noop, pprint
from bonobo.structs import Bag, Graph, Token
from bonobo.nodes import CsvReader, CsvWriter, FileReader, FileWriter, Filter, JsonReader, JsonWriter, Limit, \
PickleReader, PickleWriter, PrettyPrinter, RateLimited, Tee, arg0_to_kwargs, count, identity, kwargs_to_arg0, noop
from bonobo.strategies import create_strategy
from bonobo.structs import Bag, Graph
from bonobo.util.objects import get_name
from bonobo.io import CsvReader, CsvWriter, FileReader, FileWriter, JsonReader, JsonWriter
__all__ = []
@ -20,50 +20,57 @@ def register_api_group(*args):
@register_api
def run(graph, *chain, strategy=None, plugins=None, services=None):
def run(graph, strategy=None, plugins=None, services=None):
"""
Main entry point of bonobo. It takes a graph and creates all the necessary plumbery around to execute it.
The only necessary argument is a :class:`Graph` instance, containing the logic you actually want to execute.
By default, this graph will be executed using the "threadpool" strategy: each graph node will be wrapped in a
thread, and executed in a loop until there is no more input to this node.
You can provide plugins factory objects in the plugins list, this function will add the necessary plugins for
interactive console execution and jupyter notebook execution if it detects correctly that it runs in this context.
You'll probably want to provide a services dictionary mapping service names to service instances.
:param Graph graph: The :class:`Graph` to execute.
:param str strategy: The :class:`bonobo.strategies.base.Strategy` to use.
:param list plugins: The list of plugins to enhance execution.
:param dict services: The implementations of services this graph will use.
:return bonobo.execution.graph.GraphExecutionContext:
"""
if len(chain):
warnings.warn('DEPRECATED. You should pass a Graph instance instead of a chain.')
from bonobo import Graph
graph = Graph(graph, *chain)
strategy = create_strategy(strategy)
plugins = plugins or []
if _is_interactive_console():
from bonobo.ext.console import ConsoleOutputPlugin
if ConsoleOutputPlugin not in plugins:
plugins.append(ConsoleOutputPlugin)
from bonobo import settings
settings.check()
if _is_jupyter_notebook():
from bonobo.ext.jupyter import JupyterOutputPlugin
if JupyterOutputPlugin not in plugins:
plugins.append(JupyterOutputPlugin)
if not settings.QUIET.get(): # pragma: no cover
if _is_interactive_console():
from bonobo.ext.console import ConsoleOutputPlugin
if ConsoleOutputPlugin not in plugins:
plugins.append(ConsoleOutputPlugin)
if _is_jupyter_notebook():
try:
from bonobo.ext.jupyter import JupyterOutputPlugin
except ImportError:
logging.warning(
'Failed to load jupyter widget. Easiest way is to install the optional "jupyter" '
'dependencies with «pip install bonobo[jupyter]», but you can also install a specific '
'version by yourself.'
)
else:
if JupyterOutputPlugin not in plugins:
plugins.append(JupyterOutputPlugin)
return strategy.execute(graph, plugins=plugins, services=services)
# bonobo.structs
register_api_group(Bag, Graph)
register_api_group(Bag, Graph, Token)
# bonobo.strategies
register_api(create_strategy)
@ -71,10 +78,10 @@ register_api(create_strategy)
# Shortcut to filesystem2's open_fs, that we make available there for convenience.
@register_api
def open_fs(fs_url, *args, **kwargs):
def open_fs(fs_url=None, *args, **kwargs):
"""
Wraps :func:`fs.open_fs` function with a few candies.
:param str fs_url: A filesystem URL
:param parse_result: A parsed filesystem URL.
:type parse_result: :class:`ParseResult`
@ -85,23 +92,37 @@ def open_fs(fs_url, *args, **kwargs):
:returns: :class:`~fs.base.FS` object
"""
from fs import open_fs as _open_fs
return _open_fs(str(fs_url), *args, **kwargs)
from os.path import expanduser
from os import getcwd
if fs_url is None:
fs_url = getcwd()
return _open_fs(expanduser(str(fs_url)), *args, **kwargs)
# bonobo.basics
# bonobo.nodes
register_api_group(
CsvReader,
CsvWriter,
FileReader,
FileWriter,
Filter,
JsonReader,
JsonWriter,
Limit,
PrettyPrint,
PickleReader,
PickleWriter,
PrettyPrinter,
RateLimited,
Tee,
arg0_to_kwargs,
count,
identity,
kwargs_to_arg0,
noop,
pprint,
)
# bonobo.io
register_api_group(CsvReader, CsvWriter, FileReader, FileWriter, JsonReader, JsonWriter)
def _is_interactive_console():
import sys

2
bonobo/_version.py
View File

@ -1 +1 @@
__version__ = '0.3.0a1'
__version__ = '0.4.3'

105
bonobo/basics.py
View File

@ -1,105 +0,0 @@
import functools
from pprint import pprint as _pprint
from colorama import Fore, Style
from bonobo.config.processors import ContextProcessor
from bonobo.structs.bags import Bag
from bonobo.util.objects import ValueHolder
from bonobo.util.term import CLEAR_EOL
__all__ = [
'identity',
'Limit',
'Tee',
'count',
'pprint',
'PrettyPrint',
'noop',
]
def identity(x):
return x
def Limit(n=10):
from bonobo.constants import NOT_MODIFIED
i = 0
def _limit(*args, **kwargs):
nonlocal i, n
i += 1
if i <= n:
yield NOT_MODIFIED
_limit.__name__ = 'Limit({})'.format(n)
return _limit
def Tee(f):
from bonobo.constants import NOT_MODIFIED
@functools.wraps(f)
def wrapped(*args, **kwargs):
nonlocal f
f(*args, **kwargs)
return NOT_MODIFIED
return wrapped
def count(counter, *args, **kwargs):
counter += 1
@ContextProcessor.decorate(count)
def _count_counter(self, context):
counter = ValueHolder(0)
yield counter
context.send(Bag(counter.value))
pprint = Tee(_pprint)
def PrettyPrint(title_keys=('title', 'name', 'id'), print_values=True, sort=True):
from bonobo.constants import NOT_MODIFIED
def _pprint(*args, **kwargs):
nonlocal title_keys, sort, print_values
row = args[0]
for key in title_keys:
if key in row:
print(Style.BRIGHT, row.get(key), Style.RESET_ALL, sep='')
break
if print_values:
for k in sorted(row) if sort else row:
print(
' • ',
Fore.BLUE,
k,
Style.RESET_ALL,
' : ',
Fore.BLACK,
'(',
type(row[k]).__name__,
')',
Style.RESET_ALL,
' ',
repr(row[k]),
CLEAR_EOL,
)
yield NOT_MODIFIED
_pprint.__name__ = 'pprint'
return _pprint
def noop(*args, **kwargs): # pylint: disable=unused-argument
from bonobo.constants import NOT_MODIFIED
return NOT_MODIFIED

19
bonobo/commands/__init__.py
View File

@ -1,11 +1,13 @@
import argparse
import logging
from stevedore import ExtensionManager
from bonobo import logging, settings
logger = logging.get_logger()
def entrypoint(args=None):
parser = argparse.ArgumentParser()
parser.add_argument('--debug', '-D', action='store_true')
subparsers = parser.add_subparsers(dest='command')
subparsers.required = True
@ -17,12 +19,17 @@ def entrypoint(args=None):
parser = subparsers.add_parser(ext.name)
commands[ext.name] = ext.plugin(parser)
except Exception:
logging.exception('Error while loading command {}.'.format(ext.name))
logger.exception('Error while loading command {}.'.format(ext.name))
mgr = ExtensionManager(
namespace='bonobo.commands',
)
from stevedore import ExtensionManager
mgr = ExtensionManager(namespace='bonobo.commands')
mgr.map(register_extension)
args = parser.parse_args(args).__dict__
if args.pop('debug', False):
settings.DEBUG.set(True)
settings.LOGGING_LEVEL.set(logging.DEBUG)
logging.set_level(settings.LOGGING_LEVEL.get())
logger.debug('Command: ' + args['command'] + ' Arguments: ' + repr(args))
commands[args.pop('command')](**args)

18
bonobo/commands/init.py
View File

@ -1,16 +1,20 @@
import os
def execute():
def execute(name, branch):
try:
from edgy.project.__main__ import handle_init
from cookiecutter.main import cookiecutter
except ImportError as exc:
raise ImportError(
'You must install "edgy.project" to use this command.\n\n $ pip install edgy.project\n'
'You must install "cookiecutter" to use this command.\n\n $ pip install cookiecutter\n'
) from exc
return handle_init(os.path.join(os.getcwd(), 'Projectfile'))
return cookiecutter(
'https://github.com/python-bonobo/cookiecutter-bonobo.git',
extra_context={'name': name},
no_input=True,
checkout=branch
)
def register(parser):
parser.add_argument('name')
parser.add_argument('--branch', '-b', default='master')
return execute

98
bonobo/commands/run.py
View File

@ -3,6 +3,9 @@ import os
import bonobo
from bonobo.constants import DEFAULT_SERVICES_ATTR, DEFAULT_SERVICES_FILENAME
DEFAULT_GRAPH_FILENAMES = ('__main__.py', 'main.py',)
DEFAULT_GRAPH_ATTR = 'get_graph'
def get_default_services(filename, services=None):
dirname = os.path.dirname(filename)
@ -14,10 +17,8 @@ def get_default_services(filename, services=None):
'__name__': '__bonobo__',
'__file__': services_filename,
}
try:
exec(code, context)
except Exception:
raise
exec(code, context)
return {
**context[DEFAULT_SERVICES_ATTR](),
**(services or {}),
@ -25,26 +26,54 @@ def get_default_services(filename, services=None):
return services or {}
def read_file(file):
with file:
code = compile(file.read(), file.name, 'exec')
def _install_requirements(requirements):
"""Install requirements given a path to requirements.txt file."""
import importlib
import pip
# TODO: A few special variables should be set before running the file:
#
# See:
# - https://docs.python.org/3/reference/import.html#import-mod-attrs
# - https://docs.python.org/3/library/runpy.html#runpy.run_module
context = {
'__name__': '__bonobo__',
'__file__': file.name,
}
pip.main(['install', '-r', requirements])
# Some shenanigans to be sure everything is importable after this, especially .egg-link files which
# are referenced in *.pth files and apparently loaded by site.py at some magic bootstrap moment of the
# python interpreter.
pip.utils.pkg_resources = importlib.reload(pip.utils.pkg_resources)
import site
importlib.reload(site)
try:
exec(code, context)
except Exception as exc:
raise
graphs = dict((k, v) for k, v in context.items() if isinstance(v, bonobo.Graph))
def execute(filename, module, install=False, quiet=False, verbose=False):
import runpy
from bonobo import Graph, settings
if quiet:
settings.QUIET.set(True)
if verbose:
settings.DEBUG.set(True)
if filename:
if os.path.isdir(filename):
if install:
requirements = os.path.join(filename, 'requirements.txt')
_install_requirements(requirements)
pathname = filename
for filename in DEFAULT_GRAPH_FILENAMES:
filename = os.path.join(pathname, filename)
if os.path.exists(filename):
break
if not os.path.exists(filename):
raise IOError('Could not find entrypoint (candidates: {}).'.format(', '.join(DEFAULT_GRAPH_FILENAMES)))
elif install:
requirements = os.path.join(os.path.dirname(filename), 'requirements.txt')
_install_requirements(requirements)
context = runpy.run_path(filename, run_name='__bonobo__')
elif module:
context = runpy.run_module(module, run_name='__bonobo__')
filename = context['__file__']
else:
raise RuntimeError('UNEXPECTED: argparse should not allow this.')
graphs = dict((k, v) for k, v in context.items() if isinstance(v, Graph))
assert len(graphs) == 1, (
'Having zero or more than one graph definition in one file is unsupported for now, '
@ -54,22 +83,27 @@ def read_file(file):
graph = list(graphs.values())[0]
plugins = []
services = get_default_services(
file.name, context.get(DEFAULT_SERVICES_ATTR)() if DEFAULT_SERVICES_ATTR in context else None
filename, context.get(DEFAULT_SERVICES_ATTR)() if DEFAULT_SERVICES_ATTR in context else None
)
return graph, plugins, services
return bonobo.run(
graph,
plugins=plugins,
services=services
)
def execute(file, quiet=False):
graph, plugins, services = read_file(file)
# todo if console and not quiet, then add the console plugin
# todo when better console plugin, add it if console and just disable display
return bonobo.run(graph, plugins=plugins, services=services)
def register_generic_run_arguments(parser, required=True):
source_group = parser.add_mutually_exclusive_group(required=required)
source_group.add_argument('filename', nargs='?', type=str)
source_group.add_argument('--module', '-m', type=str)
return parser
def register(parser):
import argparse
parser.add_argument('file', type=argparse.FileType())
parser.add_argument('--quiet', action='store_true')
parser = register_generic_run_arguments(parser)
verbosity_group = parser.add_mutually_exclusive_group()
verbosity_group.add_argument('--quiet', '-q', action='store_true')
verbosity_group.add_argument('--verbose', '-v', action='store_true')
parser.add_argument('--install', '-I', action='store_true')
return execute

37
bonobo/commands/version.py
View File

@ -1,9 +1,40 @@
import bonobo
def format_version(mod, *, name=None, quiet=False):
from bonobo.util.pkgs import bonobo_packages
args = {
'name': name or mod.__name__,
'version': mod.__version__,
'location': bonobo_packages[name or mod.__name__].location
}
if not quiet:
return '{name} v.{version} (in {location})'.format(**args)
if quiet < 2:
return '{name} {version}'.format(**args)
if quiet < 3:
return '{version}'.format(**args)
raise RuntimeError('Hard to be so quiet...')
def execute():
print('{} v.{}'.format(bonobo.__name__, bonobo.__version__))
def execute(all=False, quiet=False):
import bonobo
from bonobo.util.pkgs import bonobo_packages
print(format_version(bonobo, quiet=quiet))
if all:
for name in sorted(bonobo_packages):
if name != 'bonobo':
try:
mod = __import__(name.replace('-', '_'))
try:
print(format_version(mod, name=name, quiet=quiet))
except Exception as exc:
print('{} ({})'.format(name, exc))
except ImportError as exc:
print('{} is not importable ({}).'.format(name, exc))
def register(parser):
parser.add_argument('--all', '-a', action='store_true')
parser.add_argument('--quiet', '-q', action='count')
return execute

8
bonobo/config/__init__.py
View File

@ -1,12 +1,16 @@
from bonobo.config.configurables import Configurable
from bonobo.config.options import Option
from bonobo.config.options import Method, Option
from bonobo.config.processors import ContextProcessor
from bonobo.config.services import Container, Service
from bonobo.config.services import Container, Exclusive, Service, requires
# Bonobo's Config API
__all__ = [
'Configurable',
'Container',
'ContextProcessor',
'Exclusive',
'Method',
'Option',
'Service',
'requires',
]

194
bonobo/config/configurables.py
View File

@ -1,11 +1,14 @@
from bonobo.config.processors import ContextProcessor
from bonobo.config.options import Option
from bonobo.util.inspect import isoption, iscontextprocessor
from bonobo.errors import AbstractError
from bonobo.util.collections import sortedlist
__all__ = [
'Configurable',
'Option',
]
get_creation_counter = lambda v: v._creation_counter
class ConfigurableMeta(type):
"""
@ -14,26 +17,78 @@ class ConfigurableMeta(type):
def __init__(cls, what, bases=None, dict=None):
super().__init__(what, bases, dict)
cls.__options__ = {}
cls.__positional_options__ = []
cls.__processors__ = []
cls.__processors = sortedlist()
cls.__methods = sortedlist()
cls.__options = sortedlist()
cls.__names = set()
# cls.__kwoptions = []
for typ in cls.__mro__:
for name, value in typ.__dict__.items():
if isinstance(value, Option):
if isinstance(value, ContextProcessor):
cls.__processors__.append(value)
else:
if not value.name:
value.name = name
if not name in cls.__options__:
cls.__options__[name] = value
if value.positional:
cls.__positional_options__.append(name)
for name, value in filter(lambda x: isoption(x[1]), typ.__dict__.items()):
if iscontextprocessor(value):
cls.__processors.insort((value._creation_counter, value))
continue
# This can be done before, more efficiently. Not so bad neither as this is only done at type() creation time
# (aka class Xxx(...) time) and there should not be hundreds of processors. Still not very elegant.
cls.__processors__ = sorted(cls.__processors__, key=lambda v: v._creation_counter)
if not value.name:
value.name = name
if not name in cls.__names:
cls.__names.add(name)
cls.__options.insort((not value.positional, value._creation_counter, name, value))
@property
def __options__(cls):
return ((name, option) for _, _, name, option in cls.__options)
@property
def __options_dict__(cls):
return dict(cls.__options__)
@property
def __processors__(cls):
return (processor for _, processor in cls.__processors)
def __repr__(self):
return ' '.join(('<Configurable', super(ConfigurableMeta, self).__repr__().split(' ', 1)[1], ))
try:
import _functools
except:
import functools
PartiallyConfigured = functools.partial
else:
class PartiallyConfigured(_functools.partial):
@property # TODO XXX cache this shit
def _options_values(self):
""" Simulate option values for partially configured objects. """
try:
return self.__options_values
except AttributeError:
self.__options_values = {**self.keywords}
position = 0
for name, option in self.func.__options__:
if not option.positional:
break # no positional left
if name in self.keywords:
continue # already fulfilled
self.__options_values[name] = self.args[position] if len(self.args) >= position + 1 else None
position += 1
return self.__options_values
def __getattr__(self, item):
_dict = self.func.__options_dict__
if item in _dict:
return _dict[item].__get__(self, self.func)
return getattr(self.func, item)
class Configurable(metaclass=ConfigurableMeta):
@ -43,45 +98,106 @@ class Configurable(metaclass=ConfigurableMeta):
"""
def __init__(self, *args, **kwargs):
super().__init__()
def __new__(cls, *args, _final=False, **kwargs):
"""
Custom instance builder. If not all options are fulfilled, will return a :class:`PartiallyConfigured` instance
which is just a :class:`functools.partial` object that behaves like a :class:`Configurable` instance.
# initialize option's value dictionary, used by descriptor implementation (see Option).
self.__options_values__ = {}
The special `_final` argument can be used to force final instance to be created, or an error raised if options
are missing.
:param args:
:param _final: bool
:param kwargs:
:return: Configurable or PartiallyConfigured
"""
options = tuple(cls.__options__)
# compute missing options, given the kwargs.
missing = set()
for name, option in type(self).__options__.items():
for name, option in options:
if option.required and not option.name in kwargs:
missing.add(name)
# transform positional arguments in keyword arguments if possible.
position = 0
for positional_option in self.__positional_options__:
if positional_option in missing:
kwargs[positional_option] = args[position]
position += 1
missing.remove(positional_option)
for name, option in options:
if not option.positional:
break # option orders make all positional options first, job done.
# complain if there are still missing options.
if len(missing):
raise TypeError(
'{}() missing {} required option{}: {}.'.format(
type(self).__name__,
len(missing), 's' if len(missing) > 1 else '', ', '.join(map(repr, sorted(missing)))
)
)
if not isoption(getattr(cls, name)):
missing.remove(name)
continue
if len(args) <= position:
break # no more positional arguments given.
position += 1
if name in missing:
missing.remove(name)
# complain if there is more options than possible.
extraneous = set(kwargs.keys()) - set(type(self).__options__.keys())
extraneous = set(kwargs.keys()) - (set(next(zip(*options))) if len(options) else set())
if len(extraneous):
raise TypeError(
'{}() got {} unexpected option{}: {}.'.format(
type(self).__name__,
cls.__name__,
len(extraneous), 's' if len(extraneous) > 1 else '', ', '.join(map(repr, sorted(extraneous)))
)
)
# missing options? we'll return a partial instance to finish the work later, unless we're required to be
# "final".
if len(missing):
if _final:
raise TypeError(
'{}() missing {} required option{}: {}.'.format(
cls.__name__,
len(missing), 's' if len(missing) > 1 else '', ', '.join(map(repr, sorted(missing)))
)
)
return PartiallyConfigured(cls, *args, **kwargs)
return super(Configurable, cls).__new__(cls)
def __init__(self, *args, **kwargs):
# initialize option's value dictionary, used by descriptor implementation (see Option).
self._options_values = {**kwargs}
# set option values.
for name, value in kwargs.items():
setattr(self, name, value)
position = 0
for name, option in self.__options__:
if not option.positional:
break # option orders make all positional options first
# value was overriden? Skip.
maybe_value = getattr(type(self), name)
if not isoption(maybe_value):
continue
if len(args) <= position:
break
if name in self._options_values:
raise ValueError('Already got a value for option {}'.format(name))
setattr(self, name, args[position])
position += 1
def __call__(self, *args, **kwargs):
""" You can implement a configurable callable behaviour by implemenenting the call(...) method. Of course, it is also backward compatible with legacy __call__ override.
"""
return self.call(*args, **kwargs)
@property
def __options__(self):
return type(self).__options__
@property
def __processors__(self):
return type(self).__processors__
def call(self, *args, **kwargs):
raise AbstractError('Not implemented.')

131
bonobo/config/options.py
View File

@ -1,14 +1,62 @@
from bonobo.util.inspect import istype
class Option:
"""
An Option is a descriptor for a required or optional parameter of a Configurable.
An Option is a descriptor for Configurable's parameters.
.. attribute:: type
Option type allows to provide a callable used to cast, clean or validate the option value. If not provided, or
None, the option's value will be the exact value user provided.
(default: None)
.. attribute:: required
If an option is required, an error will be raised if no value is provided (at runtime). If it is not, option
will have the default value if user does not override it at runtime.
Ignored if a default is provided, meaning that the option cannot be required.
(default: True)
.. attribute:: positional
If this is true, it'll be possible to provide the option value as a positional argument. Otherwise, it must
be provided as a keyword argument.
(default: False)
.. attribute:: default
Default value for non-required options.
(default: None)
Example:
.. code-block:: python
from bonobo.config import Configurable, Option
class Example(Configurable):
title = Option(str, required=True, positional=True)
keyword = Option(str, default='foo')
def call(self, s):
return self.title + ': ' + s + ' (' + self.keyword + ')'
example = Example('hello', keyword='bar')
"""
_creation_counter = 0
def __init__(self, type=None, *, required=False, positional=False, default=None):
def __init__(self, type=None, *, required=True, positional=False, default=None):
self.name = None
self.type = type
self.required = required
self.required = required if default is None else False
self.positional = positional
self.default = default
@ -16,13 +64,78 @@ class Option:
self._creation_counter = Option._creation_counter
Option._creation_counter += 1
def __get__(self, inst, typ):
# XXX If we call this on the type, then either return overriden value or ... ???
if inst is None:
return vars(type).get(self.name, self)
if not self.name in inst._options_values:
inst._options_values[self.name] = self.get_default()
return inst._options_values[self.name]
def __set__(self, inst, value):
inst._options_values[self.name] = self.clean(value)
def __repr__(self):
return '<{positional}{typename} {type}{name} default={default!r}{required}>'.format(
typename=type(self).__name__,
type='({})'.format(self.type) if istype(self.type) else '',
name=self.name,
positional='*' if self.positional else '**',
default=self.default,
required=' (required)' if self.required else '',
)
def clean(self, value):
return self.type(value) if self.type else value
def get_default(self):
return self.default() if callable(self.default) else self.default
def __get__(self, inst, typ):
if not self.name in inst.__options_values__:
inst.__options_values__[self.name] = self.get_default()
return inst.__options_values__[self.name]
class Method(Option):
"""
A Method is a special callable-valued option, that can be used in three different ways (but for same purpose).
* Like a normal option, the value can be provided to the Configurable constructor.
>>> from bonobo.config import Configurable, Method
>>> class MethodExample(Configurable):
... handler = Method()
>>> example1 = MethodExample(handler=str.upper)
* It can be used by a child class that overrides the Method with a normal method.
>>> class ChildMethodExample(MethodExample):
... def handler(self, s: str):
... return s.upper()
>>> example2 = ChildMethodExample()
* Finally, it also enables the class to be used as a decorator, to generate a subclass providing the Method a value.
>>> @MethodExample
... def OtherChildMethodExample(s):
... return s.upper()
>>> example3 = OtherChildMethodExample()
"""
def __init__(self, *, required=True, positional=True):
super().__init__(None, required=required, positional=positional)
def __set__(self, inst, value):
inst.__options_values__[self.name] = self.type(value) if self.type else value
if not hasattr(value, '__call__'):
raise TypeError(
'Option of type {!r} is expecting a callable value, got {!r} object (which is not).'.
format(type(self).__name__, type(value).__name__)
)
inst._options_values[self.name] = self.type(value) if self.type else value
def __call__(self, *args, **kwargs):
# only here to trick IDEs into thinking this is callable.
raise NotImplementedError('You cannot call the descriptor')

112
bonobo/config/processors.py
View File

@ -1,16 +1,42 @@
import functools
import types
from bonobo.util.compat import deprecated_alias, deprecated
from collections import Iterable
from contextlib import contextmanager
from bonobo.config.options import Option
from bonobo.util.compat import deprecated_alias
from bonobo.util.iterators import ensure_tuple
_CONTEXT_PROCESSORS_ATTR = '__processors__'
class ContextProcessor(Option):
"""
A ContextProcessor is a kind of transformation decorator that can setup and teardown a transformation and runtime
related dependencies, at the execution level.
It works like a yielding context manager, and is the recommended way to setup and teardown objects you'll need
in the context of one execution. It's the way to overcome the stateless nature of transformations.
The yielded values will be passed as positional arguments to the next context processors (order do matter), and
finally to the __call__ method of the transformation.
Warning: this may change for a similar but simpler implementation, don't relly too much on it (yet).
Example:
>>> from bonobo.config import Configurable
>>> from bonobo.util.objects import ValueHolder
>>> class Counter(Configurable):
... @ContextProcessor
... def counter(self, context):
... yield ValueHolder(0)
...
... def __call__(self, counter, *args, **kwargs):
... counter += 1
... yield counter.get()
"""
@property
def __name__(self):
return self.func.__name__
@ -48,82 +74,66 @@ class ContextCurrifier:
def __init__(self, wrapped, *initial_context):
self.wrapped = wrapped
self.context = tuple(initial_context)
self._stack = []
self._stack, self._stack_values = None, None
def __iter__(self):
yield from self.wrapped
def __call__(self, *args, **kwargs):
if not callable(self.wrapped) and isinstance(self.wrapped, Iterable):
return self.__iter__()
return self.wrapped(*self.context, *args, **kwargs)
def setup(self, *context):
if len(self._stack):
if self._stack is not None:
raise RuntimeError('Cannot setup context currification twice.')
self._stack, self._stack_values = list(), list()
for processor in resolve_processors(self.wrapped):
_processed = processor(self.wrapped, *context, *self.context)
_append_to_context = next(_processed)
self._stack_values.append(_append_to_context)
if _append_to_context is not None:
self.context += ensure_tuple(_append_to_context)
self._stack.append(_processed)
def __call__(self, *args, **kwargs):
return self.wrapped(*self.context, *args, **kwargs)
def teardown(self):
while len(self._stack):
while self._stack:
processor = self._stack.pop()
try:
# todo yield from ? how to ?
next(processor)
processor.send(self._stack_values.pop())
except StopIteration as exc:
# This is normal, and wanted.
pass
else:
# No error ? We should have had StopIteration ...
raise RuntimeError('Context processors should not yield more than once.')
self._stack, self._stack_values = None, None
@contextmanager
def as_contextmanager(self, *context):
"""
Convenience method to use it as a contextmanager, mostly for test purposes.
@deprecated
def add_context_processor(cls_or_func, context_processor):
getattr(cls_or_func, _CONTEXT_PROCESSORS_ATTR).append(context_processor)
Example:
>>> with ContextCurrifier(node).as_contextmanager(context) as stack:
... stack()
@deprecated
def contextual(cls_or_func):
"""
Make sure an element has the context processors collection.
:param cls_or_func:
"""
if not add_context_processor.__name__ in cls_or_func.__dict__:
setattr(cls_or_func, add_context_processor.__name__, functools.partial(add_context_processor, cls_or_func))
if isinstance(cls_or_func, types.FunctionType):
try:
getattr(cls_or_func, _CONTEXT_PROCESSORS_ATTR)
except AttributeError:
setattr(cls_or_func, _CONTEXT_PROCESSORS_ATTR, [])
return cls_or_func
if not _CONTEXT_PROCESSORS_ATTR in cls_or_func.__dict__:
setattr(cls_or_func, _CONTEXT_PROCESSORS_ATTR, [])
_processors = getattr(cls_or_func, _CONTEXT_PROCESSORS_ATTR)
for processor in cls_or_func.__dict__.values():
if isinstance(processor, ContextProcessor):
_processors.append(processor)
# This is needed for python 3.5, python 3.6 should be fine, but it's considered an implementation detail.
_processors.sort(key=lambda proc: proc._creation_counter)
return cls_or_func
:param context:
:return:
"""
self.setup(*context)
yield self
self.teardown()
def resolve_processors(mixed):
try:
yield from mixed.__processors__
except AttributeError:
# old code, deprecated usage
if isinstance(mixed, types.FunctionType):
yield from getattr(mixed, _CONTEXT_PROCESSORS_ATTR, ())
for cls in reversed((mixed if isinstance(mixed, type) else type(mixed)).__mro__):
yield from cls.__dict__.get(_CONTEXT_PROCESSORS_ATTR, ())
return ()
yield from ()
get_context_processors = deprecated_alias('get_context_processors', resolve_processors)

75
bonobo/config/services.py
View File

@ -1,7 +1,10 @@
import re
import threading
import types
from contextlib import ContextDecorator
from bonobo.config.options import Option
from bonobo.errors import MissingServiceImplementationError
_service_name_re = re.compile(r"^[^\d\W]\w*(:?\.[^\d\W]\w*)*$", re.UNICODE)
@ -39,6 +42,10 @@ class Service(Option):
The main goal is not to tie transformations to actual dependencies, so the same can be run in different contexts
(stages like preprod, prod, or tenants like client1, client2, or anything you want).
.. attribute:: name
Service name will be used to retrieve the implementation at runtime.
"""
@ -46,10 +53,14 @@ class Service(Option):
super().__init__(str, required=False, default=name)
def __set__(self, inst, value):
inst.__options_values__[self.name] = validate_service_name(value)
inst._options_values[self.name] = validate_service_name(value)
def resolve(self, inst, services):
return services.get(getattr(inst, self.name))
try:
name = getattr(inst, self.name)
except AttributeError:
name = self.name
return services.get(name)
class Container(dict):
@ -64,7 +75,7 @@ class Container(dict):
def args_for(self, mixed):
try:
options = mixed.__options__
options = dict(mixed.__options__)
except AttributeError:
options = {}
@ -74,8 +85,64 @@ class Container(dict):
if not name in self:
if default:
return default
raise KeyError('Cannot resolve service {!r} using provided service collection.'.format(name))
raise MissingServiceImplementationError(
'Cannot resolve service {!r} using provided service collection.'.format(name)
)
value = super().get(name)
# XXX this is not documented and can lead to errors.
if isinstance(value, types.LambdaType):
value = value(self)
return value
class Exclusive(ContextDecorator):
"""
Decorator and context manager used to require exclusive usage of an object, most probably a service. It's usefull
for example if call order matters on a service implementation (think of an http api that requires a nonce or version
parameter ...).
Usage:
>>> def handler(some_service):
... with Exclusive(some_service):
... some_service.call_1()
... some_service.call_2()
... some_service.call_3()
This will ensure that nobody else is using the same service while in the "with" block, using a lock primitive to
ensure that.
"""
_locks = {}
def __init__(self, wrapped):
self._wrapped = wrapped
def get_lock(self):
_id = id(self._wrapped)
if not _id in Exclusive._locks:
Exclusive._locks[_id] = threading.RLock()
return Exclusive._locks[_id]
def __enter__(self):
self.get_lock().acquire()
return self._wrapped
def __exit__(self, *exc):
self.get_lock().release()
def requires(*service_names):
def decorate(mixed):
try:
options = mixed.__options__
except AttributeError:
mixed.__options__ = options = {}
for service_name in service_names:
service = Service(service_name)
service.name = service_name
options[service_name] = service
return mixed
return decorate

1
bonobo/core/__init__.py
View File

@ -1 +0,0 @@
""" Core required libraries. """

25
bonobo/errors.py
View File

@ -52,3 +52,28 @@ class ValidationError(RuntimeError):
class ProhibitedOperationError(RuntimeError):
pass
class ConfigurationError(Exception):
pass
class UnrecoverableError(Exception):
"""Flag for errors that must interrupt the workflow, either because they will happen for sure on each node run, or
because you know that your transformation has no point continuing runnning after a bad event."""
class UnrecoverableValueError(UnrecoverableError, ValueError):
pass
class UnrecoverableRuntimeError(UnrecoverableError, RuntimeError):
pass
class UnrecoverableNotImplementedError(UnrecoverableError, NotImplementedError):
pass
class MissingServiceImplementationError(UnrecoverableError, KeyError):
pass

18
bonobo/examples/__init__.py
View File

@ -5,9 +5,19 @@ def require(package, requirement=None):
return __import__(package)
except ImportError:
from colorama import Fore, Style
print(Fore.YELLOW, 'This example requires the {!r} package. Install it using:'.format(requirement),
Style.RESET_ALL, sep='')
print(
Fore.YELLOW,
'This example requires the {!r} package. Install it using:'.
format(requirement),
Style.RESET_ALL,
sep=''
)
print()
print(Fore.YELLOW, ' $ pip install {!s}'.format(requirement), Style.RESET_ALL, sep='')
print(
Fore.YELLOW,
' $ pip install {!s}'.format(requirement),
Style.RESET_ALL,
sep=''
)
print()
raise
raise

204
bonobo/examples/datasets/coffeeshops.json
View File

@ -1,37 +1,40 @@
{"Ext\u00e9rieur Quai": "5, rue d'Alsace, 75010 Paris, France",
"Le Sully": "6 Bd henri IV, 75004 Paris, France",
{"les montparnos": "65 boulevard Pasteur, 75015 Paris, France",
"Coffee Chope": "344Vrue Vaugirard, 75015 Paris, France",
"Caf\u00e9 Lea": "5 rue Claude Bernard, 75005 Paris, France",
"Le Bellerive": "71 quai de Seine, 75019 Paris, France",
"Le drapeau de la fidelit\u00e9": "21 rue Copreaux, 75015 Paris, France",
"O q de poule": "53 rue du ruisseau, 75018 Paris, France",
"Le Pas Sage": "1 Passage du Grand Cerf, 75002 Paris, France",
"La Renaissance": "112 Rue Championnet, 75018 Paris, France",
"Le caf\u00e9 des amis": "125 rue Blomet, 75015 Paris, France",
"Le chantereine": "51 Rue Victoire, 75009 Paris, France",
"Le M\u00fcller": "11 rue Feutrier, 75018 Paris, France",
"Le drapeau de la fidelit\u00e9": "21 rue Copreaux, 75015 Paris, France",
"Le caf\u00e9 des amis": "125 rue Blomet, 75015 Paris, France",
"Le Caf\u00e9 Livres": "10 rue Saint Martin, 75004 Paris, France",
"Le Bosquet": "46 avenue Bosquet, 75007 Paris, France",
"Le Brio": "216, rue Marcadet, 75018 Paris, France",
"Le Kleemend's": "34 avenue Pierre Mend\u00e8s-France, 75013 Paris, France",
"Caf\u00e9 Pierre": "202 rue du faubourg st antoine, 75012 Paris, France",
"Les Arcades": "61 rue de Ponthieu, 75008 Paris, France",
"Le Square": "31 rue Saint-Dominique, 75007 Paris, France",
"Assaporare Dix sur Dix": "75, avenue Ledru-Rollin, 75012 Paris, France",
"Au cerceau d'or": "129 boulevard sebastopol, 75002 Paris, France",
"Aux cadrans": "21 ter boulevard Diderot, 75012 Paris, France",
"Caf\u00e9 antoine": "17 rue Jean de la Fontaine, 75016 Paris, France",
"Caf\u00e9 Lea": "5 rue Claude Bernard, 75005 Paris, France",
"Cardinal Saint-Germain": "11 boulevard Saint-Germain, 75005 Paris, France",
"D\u00e9d\u00e9 la frite": "52 rue Notre-Dame des Victoires, 75002 Paris, France",
"Ext\u00e9rieur Quai": "5, rue d'Alsace, 75010 Paris, France",
"La Bauloise": "36 rue du hameau, 75015 Paris, France",
"Le Bellerive": "71 quai de Seine, 75019 Paris, France",
"Le bistrot de Ma\u00eblle et Augustin": "42 rue coquill\u00e8re, 75001 Paris, France",
"Le Dellac": "14 rue Rougemont, 75009 Paris, France",
"Le Bosquet": "46 avenue Bosquet, 75007 Paris, France",
"Le Sully": "6 Bd henri IV, 75004 Paris, France",
"Le Felteu": "1 rue Pecquay, 75004 Paris, France",
"Le bistrot de Ma\u00eblle et Augustin": "42 rue coquill\u00e8re, 75001 Paris, France",
"D\u00e9d\u00e9 la frite": "52 rue Notre-Dame des Victoires, 75002 Paris, France",
"Cardinal Saint-Germain": "11 boulevard Saint-Germain, 75005 Paris, France",
"Le Reynou": "2 bis quai de la m\u00e9gisserie, 75001 Paris, France",
"Aux cadrans": "21 ter boulevard Diderot, 75012 Paris, France",
"Le Saint Jean": "23 rue des abbesses, 75018 Paris, France",
"les montparnos": "65 boulevard Pasteur, 75015 Paris, France",
"La Renaissance": "112 Rue Championnet, 75018 Paris, France",
"Le Square": "31 rue Saint-Dominique, 75007 Paris, France",
"Les Arcades": "61 rue de Ponthieu, 75008 Paris, France",
"Le Kleemend's": "34 avenue Pierre Mend\u00e8s-France, 75013 Paris, France",
"Assaporare Dix sur Dix": "75, avenue Ledru-Rollin, 75012 Paris, France",
"Caf\u00e9 Pierre": "202 rue du faubourg st antoine, 75012 Paris, France",
"Caf\u00e9 antoine": "17 rue Jean de la Fontaine, 75016 Paris, France",
"Au cerceau d'or": "129 boulevard sebastopol, 75002 Paris, France",
"La Caravane": "Rue de la Fontaine au Roi, 75011 Paris, France",
"Le Pas Sage": "1 Passage du Grand Cerf, 75002 Paris, France",
"Le Caf\u00e9 Livres": "10 rue Saint Martin, 75004 Paris, France",
"Le Chaumontois": "12 rue Armand Carrel, 75018 Paris, France",
"Drole d'endroit pour une rencontre": "58 rue de Montorgueil, 75002 Paris, France",
"Le pari's caf\u00e9": "104 rue caulaincourt, 75018 Paris, France",
"Le Poulailler": "60 rue saint-sabin, 75011 Paris, France",
"Chai 33": "33 Cour Saint Emilion, 75012 Paris, France",
"L'Assassin": "99 rue Jean-Pierre Timbaud, 75011 Paris, France",
"l'Usine": "1 rue d'Avron, 75020 Paris, France",
"La Bricole": "52 rue Liebniz, 75018 Paris, France",
@ -84,99 +87,96 @@
"Le Ragueneau": "202 rue Saint-Honor\u00e9, 75001 Paris, France",
"Le refuge": "72 rue lamarck, 75018 Paris, France",
"Le sully": "13 rue du Faubourg Saint Denis, 75010 Paris, France",
"L'antre d'eux": "16 rue DE MEZIERES, 75006 Paris, France",
"Le bal du pirate": "60 rue des bergers, 75015 Paris, France",
"zic zinc": "95 rue claude decaen, 75012 Paris, France",
"l'orillon bar": "35 rue de l'orillon, 75011 Paris, France",
"Le Zazabar": "116 Rue de M\u00e9nilmontant, 75020 Paris, France",
"L'In\u00e9vitable": "22 rue Linn\u00e9, 75005 Paris, France",
"Le Dunois": "77 rue Dunois, 75013 Paris, France",
"Ragueneau": "202 rue Saint Honor\u00e9, 75001 Paris, France",
"Le Caminito": "48 rue du Dessous des Berges, 75013 Paris, France",
"Epicerie Musicale": "55bis quai de Valmy, 75010 Paris, France",
"Le petit Bretonneau": "Le petit Bretonneau - \u00e0 l'int\u00e9rieur de l'H\u00f4pital, 75018 Paris, France",
"Le Centenaire": "104 rue amelot, 75011 Paris, France",
"La Montagne Sans Genevi\u00e8ve": "13 Rue du Pot de Fer, 75005 Paris, France",
"Les P\u00e8res Populaires": "46 rue de Buzenval, 75020 Paris, France",
"Cafe de grenelle": "188 rue de Grenelle, 75007 Paris, France",
"Le relais de la victoire": "73 rue de la Victoire, 75009 Paris, France",
"Le Caminito": "48 rue du Dessous des Berges, 75013 Paris, France",
"Le petit Bretonneau": "Le petit Bretonneau - \u00e0 l'int\u00e9rieur de l'H\u00f4pital, 75018 Paris, France",
"La chaumi\u00e8re gourmande": "Route de la Muette \u00e0 Neuilly",
"Club hippique du Jardin d\u2019Acclimatation": "75016 Paris, France",
"Le Chaumontois": "12 rue Armand Carrel, 75018 Paris, France",
"Caves populaires": "22 rue des Dames, 75017 Paris, France",
"Caprice caf\u00e9": "12 avenue Jean Moulin, 75014 Paris, France",
"Tamm Bara": "7 rue Clisson, 75013 Paris, France",
"L'anjou": "1 rue de Montholon, 75009 Paris, France",
"Caf\u00e9 dans l'aerogare Air France Invalides": "2 rue Robert Esnault Pelterie, 75007 Paris, France",
"Chez Prune": "36 rue Beaurepaire, 75010 Paris, France",
"Au Vin Des Rues": "21 rue Boulard, 75014 Paris, France",
"bistrot les timbr\u00e9s": "14 rue d'alleray, 75015 Paris, France",
"Caf\u00e9 beauveau": "9 rue de Miromesnil, 75008 Paris, France",
"Le bal du pirate": "60 rue des bergers, 75015 Paris, France",
"Le Zazabar": "116 Rue de M\u00e9nilmontant, 75020 Paris, France",
"L'antre d'eux": "16 rue DE MEZIERES, 75006 Paris, France",
"l'orillon bar": "35 rue de l'orillon, 75011 Paris, France",
"zic zinc": "95 rue claude decaen, 75012 Paris, France",
"Les P\u00e8res Populaires": "46 rue de Buzenval, 75020 Paris, France",
"Epicerie Musicale": "55bis quai de Valmy, 75010 Paris, France",
"Le relais de la victoire": "73 rue de la Victoire, 75009 Paris, France",
"Le Centenaire": "104 rue amelot, 75011 Paris, France",
"Cafe de grenelle": "188 rue de Grenelle, 75007 Paris, France",
"Ragueneau": "202 rue Saint Honor\u00e9, 75001 Paris, France",
"Caf\u00e9 Pistache": "9 rue des petits champs, 75001 Paris, France",
"La Cagnotte": "13 Rue Jean-Baptiste Dumay, 75020 Paris, France",
"le 1 cinq": "172 rue de vaugirard, 75015 Paris, France",
"Le Killy Jen": "28 bis boulevard Diderot, 75012 Paris, France",
"Caf\u00e9 beauveau": "9 rue de Miromesnil, 75008 Paris, France",
"le 1 cinq": "172 rue de vaugirard, 75015 Paris, France",
"Les Artisans": "106 rue Lecourbe, 75015 Paris, France",
"Peperoni": "83 avenue de Wagram, 75001 Paris, France",
"le lutece": "380 rue de vaugirard, 75015 Paris, France",
"Brasiloja": "16 rue Ganneron, 75018 Paris, France",
"Rivolux": "16 rue de Rivoli, 75004 Paris, France",
"Chai 33": "33 Cour Saint Emilion, 75012 Paris, France",
"L'europ\u00e9en": "21 Bis Boulevard Diderot, 75012 Paris, France",
"NoMa": "39 rue Notre Dame de Nazareth, 75003 Paris, France",
"O'Paris": "1 Rue des Envierges, 75020 Paris, France",
"Caf\u00e9 Clochette": "16 avenue Richerand, 75010 Paris, France",
"La cantoche de Paname": "40 Boulevard Beaumarchais, 75011 Paris, France",
"Le Saint Ren\u00e9": "148 Boulevard de Charonne, 75020 Paris, France",
"La Libert\u00e9": "196 rue du faubourg saint-antoine, 75012 Paris, France",
"Chez Rutabaga": "16 rue des Petits Champs, 75002 Paris, France",
"Le BB (Bouchon des Batignolles)": "2 rue Lemercier, 75017 Paris, France",
"La Brocante": "10 rue Rossini, 75009 Paris, France",
"Le Plomb du cantal": "3 rue Ga\u00eet\u00e9, 75014 Paris, France",
"Les caves populaires": "22 rue des Dames, 75017 Paris, France",
"Chez Luna": "108 rue de M\u00e9nilmontant, 75020 Paris, France",
"Le bar Fleuri": "1 rue du Plateau, 75019 Paris, France",
"Trois pi\u00e8ces cuisine": "101 rue des dames, 75017 Paris, France",
"Le Zinc": "61 avenue de la Motte Picquet, 75015 Paris, France",
"La cantine de Zo\u00e9": "136 rue du Faubourg poissonni\u00e8re, 75010 Paris, France",
"Les Vendangeurs": "6/8 rue Stanislas, 75006 Paris, France",
"L'avant comptoir": "3 carrefour de l'Od\u00e9on, 75006 Paris, France",
"Le Brio": "216, rue Marcadet, 75018 Paris, France",
"Tamm Bara": "7 rue Clisson, 75013 Paris, France",
"Caf\u00e9 dans l'aerogare Air France Invalides": "2 rue Robert Esnault Pelterie, 75007 Paris, France",
"bistrot les timbr\u00e9s": "14 rue d'alleray, 75015 Paris, France",
"Caprice caf\u00e9": "12 avenue Jean Moulin, 75014 Paris, France",
"Caves populaires": "22 rue des Dames, 75017 Paris, France",
"Au Vin Des Rues": "21 rue Boulard, 75014 Paris, France",
"Chez Prune": "36 rue Beaurepaire, 75010 Paris, France",
"L'In\u00e9vitable": "22 rue Linn\u00e9, 75005 Paris, France",
"L'anjou": "1 rue de Montholon, 75009 Paris, France",
"Botak cafe": "1 rue Paul albert, 75018 Paris, France",
"le chateau d'eau": "67 rue du Ch\u00e2teau d'eau, 75010 Paris, France",
"Bistrot Saint-Antoine": "58 rue du Fbg Saint-Antoine, 75012 Paris, France",
"Chez Oscar": "11/13 boulevard Beaumarchais, 75004 Paris, France",
"Le Fronton": "63 rue de Ponthieu, 75008 Paris, France",
"Le Piquet": "48 avenue de la Motte Picquet, 75015 Paris, France",
"Le Tournebride": "104 rue Mouffetard, 75005 Paris, France",
"L'avant comptoir": "3 carrefour de l'Od\u00e9on, 75006 Paris, France",
"le chateau d'eau": "67 rue du Ch\u00e2teau d'eau, 75010 Paris, France",
"Les Vendangeurs": "6/8 rue Stanislas, 75006 Paris, France",
"maison du vin": "52 rue des plantes, 75014 Paris, France",
"Coffee Chope": "344Vrue Vaugirard, 75015 Paris, France",
"L'entrep\u00f4t": "157 rue Bercy 75012 Paris, 75012 Paris, France",
"Le Tournebride": "104 rue Mouffetard, 75005 Paris, France",
"Le Fronton": "63 rue de Ponthieu, 75008 Paris, France",
"Le BB (Bouchon des Batignolles)": "2 rue Lemercier, 75017 Paris, France",
"La cantine de Zo\u00e9": "136 rue du Faubourg poissonni\u00e8re, 75010 Paris, France",
"Chez Rutabaga": "16 rue des Petits Champs, 75002 Paris, France",
"Les caves populaires": "22 rue des Dames, 75017 Paris, France",
"Le Plomb du cantal": "3 rue Ga\u00eet\u00e9, 75014 Paris, France",
"Trois pi\u00e8ces cuisine": "101 rue des dames, 75017 Paris, France",
"La Brocante": "10 rue Rossini, 75009 Paris, France",
"Le Zinc": "61 avenue de la Motte Picquet, 75015 Paris, France",
"Chez Luna": "108 rue de M\u00e9nilmontant, 75020 Paris, France",
"Le bar Fleuri": "1 rue du Plateau, 75019 Paris, France",
"La Libert\u00e9": "196 rue du faubourg saint-antoine, 75012 Paris, France",
"La cantoche de Paname": "40 Boulevard Beaumarchais, 75011 Paris, France",
"Le Saint Ren\u00e9": "148 Boulevard de Charonne, 75020 Paris, France",
"Caf\u00e9 Clochette": "16 avenue Richerand, 75010 Paris, France",
"L'europ\u00e9en": "21 Bis Boulevard Diderot, 75012 Paris, France",
"NoMa": "39 rue Notre Dame de Nazareth, 75003 Paris, France",
"le lutece": "380 rue de vaugirard, 75015 Paris, France",
"O'Paris": "1 Rue des Envierges, 75020 Paris, France",
"Rivolux": "16 rue de Rivoli, 75004 Paris, France",
"Brasiloja": "16 rue Ganneron, 75018 Paris, France",
"Institut des Cultures d'Islam": "19-23 rue L\u00e9on, 75018 Paris, France",
"Canopy Caf\u00e9 associatif": "19 rue Pajol, 75018 Paris, France",
"Petits Freres des Pauvres": "47 rue de Batignolles, 75017 Paris, France",
"Le Lucernaire": "53 rue Notre-Dame des Champs, 75006 Paris, France",
"L'Angle": "28 rue de Ponthieu, 75008 Paris, France",
"Le Caf\u00e9 d'avant": "35 rue Claude Bernard, 75005 Paris, France",
"Caf\u00e9 Dupont": "198 rue de la Convention, 75015 Paris, France",
"Le S\u00e9vign\u00e9": "15 rue du Parc Royal, 75003 Paris, France",
"L'Entracte": "place de l'opera, 75002 Paris, France",
"Panem": "18 rue de Crussol, 75011 Paris, France",
"Au pays de Vannes": "34 bis rue de Wattignies, 75012 Paris, France",
"l'El\u00e9phant du nil": "125 Rue Saint-Antoine, 75004 Paris, France",
"L'\u00e2ge d'or": "26 rue du Docteur Magnan, 75013 Paris, France",
"Le Comptoir": "354 bis rue Vaugirard, 75015 Paris, France",
"L'horizon": "93, rue de la Roquette, 75011 Paris, France",
"L'empreinte": "54, avenue Daumesnil, 75012 Paris, France",
"Caf\u00e9 Victor": "10 boulevard Victor, 75015 Paris, France",
"Caf\u00e9 Varenne": "36 rue de Varenne, 75007 Paris, France",
"Le Brigadier": "12 rue Blanche, 75009 Paris, France",
"Waikiki": "10 rue d\"Ulm, 75005 Paris, France",
"Le Parc Vaugirard": "358 rue de Vaugirard, 75015 Paris, France",
"Pari's Caf\u00e9": "174 avenue de Clichy, 75017 Paris, France",
"Melting Pot": "3 rue de Lagny, 75020 Paris, France",
"le Zango": "58 rue Daguerre, 75014 Paris, France",
"Chez Miamophile": "6 rue M\u00e9lingue, 75019 Paris, France",
"Le caf\u00e9 Monde et M\u00e9dias": "Place de la R\u00e9publique, 75003 Paris, France",
"Caf\u00e9 rallye tournelles": "11 Quai de la Tournelle, 75005 Paris, France",
"Brasserie le Morvan": "61 rue du ch\u00e2teau d'eau, 75010 Paris, France",
"Chez Miamophile": "6 rue M\u00e9lingue, 75019 Paris, France",
"La Caravane": "Rue de la Fontaine au Roi, 75011 Paris, France",
"Panem": "18 rue de Crussol, 75011 Paris, France",
"Petits Freres des Pauvres": "47 rue de Batignolles, 75017 Paris, France",
"Caf\u00e9 Dupont": "198 rue de la Convention, 75015 Paris, France",
"L'Angle": "28 rue de Ponthieu, 75008 Paris, France",
"Institut des Cultures d'Islam": "19-23 rue L\u00e9on, 75018 Paris, France",
"Canopy Caf\u00e9 associatif": "19 rue Pajol, 75018 Paris, France",
"L'Entracte": "place de l'opera, 75002 Paris, France",
"Le S\u00e9vign\u00e9": "15 rue du Parc Royal, 75003 Paris, France",
"Le Caf\u00e9 d'avant": "35 rue Claude Bernard, 75005 Paris, France",
"Le Lucernaire": "53 rue Notre-Dame des Champs, 75006 Paris, France",
"Le Brigadier": "12 rue Blanche, 75009 Paris, France",
"L'\u00e2ge d'or": "26 rue du Docteur Magnan, 75013 Paris, France",
"Caf\u00e9 Victor": "10 boulevard Victor, 75015 Paris, France",
"L'empreinte": "54, avenue Daumesnil, 75012 Paris, France",
"L'horizon": "93, rue de la Roquette, 75011 Paris, France",
"Waikiki": "10 rue d\"Ulm, 75005 Paris, France",
"Au pays de Vannes": "34 bis rue de Wattignies, 75012 Paris, France",
"Caf\u00e9 Varenne": "36 rue de Varenne, 75007 Paris, France",
"l'El\u00e9phant du nil": "125 Rue Saint-Antoine, 75004 Paris, France",
"Le Comptoir": "354 bis rue Vaugirard, 75015 Paris, France",
"Le Parc Vaugirard": "358 rue de Vaugirard, 75015 Paris, France",
"le Zango": "58 rue Daguerre, 75014 Paris, France",
"Melting Pot": "3 rue de Lagny, 75020 Paris, France",
"Pari's Caf\u00e9": "174 avenue de Clichy, 75017 Paris, France"}
"L'entrep\u00f4t": "157 rue Bercy 75012 Paris, 75012 Paris, France"}

204
bonobo/examples/datasets/coffeeshops.txt
View File

@ -1,37 +1,40 @@
Extérieur Quai, 5, rue d'Alsace, 75010 Paris, France
Le Sully, 6 Bd henri IV, 75004 Paris, France
les montparnos, 65 boulevard Pasteur, 75015 Paris, France
Coffee Chope, 344Vrue Vaugirard, 75015 Paris, France
Café Lea, 5 rue Claude Bernard, 75005 Paris, France
Le Bellerive, 71 quai de Seine, 75019 Paris, France
Le drapeau de la fidelité, 21 rue Copreaux, 75015 Paris, France
O q de poule, 53 rue du ruisseau, 75018 Paris, France
Le Pas Sage, 1 Passage du Grand Cerf, 75002 Paris, France
La Renaissance, 112 Rue Championnet, 75018 Paris, France
Le café des amis, 125 rue Blomet, 75015 Paris, France
Le chantereine, 51 Rue Victoire, 75009 Paris, France
Le Müller, 11 rue Feutrier, 75018 Paris, France
Le drapeau de la fidelité, 21 rue Copreaux, 75015 Paris, France
Le café des amis, 125 rue Blomet, 75015 Paris, France
Le Café Livres, 10 rue Saint Martin, 75004 Paris, France
Le Bosquet, 46 avenue Bosquet, 75007 Paris, France
Le Brio, 216, rue Marcadet, 75018 Paris, France
Le Kleemend's, 34 avenue Pierre Mendès-France, 75013 Paris, France
Café Pierre, 202 rue du faubourg st antoine, 75012 Paris, France
Les Arcades, 61 rue de Ponthieu, 75008 Paris, France
Le Square, 31 rue Saint-Dominique, 75007 Paris, France
Assaporare Dix sur Dix, 75, avenue Ledru-Rollin, 75012 Paris, France
Au cerceau d'or, 129 boulevard sebastopol, 75002 Paris, France
Aux cadrans, 21 ter boulevard Diderot, 75012 Paris, France
Café antoine, 17 rue Jean de la Fontaine, 75016 Paris, France
Café Lea, 5 rue Claude Bernard, 75005 Paris, France
Cardinal Saint-Germain, 11 boulevard Saint-Germain, 75005 Paris, France
Dédé la frite, 52 rue Notre-Dame des Victoires, 75002 Paris, France
Extérieur Quai, 5, rue d'Alsace, 75010 Paris, France
La Bauloise, 36 rue du hameau, 75015 Paris, France
Le Bellerive, 71 quai de Seine, 75019 Paris, France
Le bistrot de Maëlle et Augustin, 42 rue coquillère, 75001 Paris, France
Le Dellac, 14 rue Rougemont, 75009 Paris, France
Le Bosquet, 46 avenue Bosquet, 75007 Paris, France
Le Sully, 6 Bd henri IV, 75004 Paris, France
Le Felteu, 1 rue Pecquay, 75004 Paris, France
Le bistrot de Maëlle et Augustin, 42 rue coquillère, 75001 Paris, France
Dédé la frite, 52 rue Notre-Dame des Victoires, 75002 Paris, France
Cardinal Saint-Germain, 11 boulevard Saint-Germain, 75005 Paris, France
Le Reynou, 2 bis quai de la mégisserie, 75001 Paris, France
Aux cadrans, 21 ter boulevard Diderot, 75012 Paris, France
Le Saint Jean, 23 rue des abbesses, 75018 Paris, France
les montparnos, 65 boulevard Pasteur, 75015 Paris, France
La Renaissance, 112 Rue Championnet, 75018 Paris, France
Le Square, 31 rue Saint-Dominique, 75007 Paris, France
Les Arcades, 61 rue de Ponthieu, 75008 Paris, France
Le Kleemend's, 34 avenue Pierre Mendès-France, 75013 Paris, France
Assaporare Dix sur Dix, 75, avenue Ledru-Rollin, 75012 Paris, France
Café Pierre, 202 rue du faubourg st antoine, 75012 Paris, France
Café antoine, 17 rue Jean de la Fontaine, 75016 Paris, France
Au cerceau d'or, 129 boulevard sebastopol, 75002 Paris, France
La Caravane, Rue de la Fontaine au Roi, 75011 Paris, France
Le Pas Sage, 1 Passage du Grand Cerf, 75002 Paris, France
Le Café Livres, 10 rue Saint Martin, 75004 Paris, France
Le Chaumontois, 12 rue Armand Carrel, 75018 Paris, France
Drole d'endroit pour une rencontre, 58 rue de Montorgueil, 75002 Paris, France
Le pari's café, 104 rue caulaincourt, 75018 Paris, France
Le Poulailler, 60 rue saint-sabin, 75011 Paris, France
Chai 33, 33 Cour Saint Emilion, 75012 Paris, France
L'Assassin, 99 rue Jean-Pierre Timbaud, 75011 Paris, France
l'Usine, 1 rue d'Avron, 75020 Paris, France
La Bricole, 52 rue Liebniz, 75018 Paris, France
@ -84,99 +87,96 @@ Le Germinal, 95 avenue Emile Zola, 75015 Paris, France
Le Ragueneau, 202 rue Saint-Honoré, 75001 Paris, France
Le refuge, 72 rue lamarck, 75018 Paris, France
Le sully, 13 rue du Faubourg Saint Denis, 75010 Paris, France
L'antre d'eux, 16 rue DE MEZIERES, 75006 Paris, France
Le bal du pirate, 60 rue des bergers, 75015 Paris, France
zic zinc, 95 rue claude decaen, 75012 Paris, France
l'orillon bar, 35 rue de l'orillon, 75011 Paris, France
Le Zazabar, 116 Rue de Ménilmontant, 75020 Paris, France
L'Inévitable, 22 rue Linné, 75005 Paris, France
Le Dunois, 77 rue Dunois, 75013 Paris, France
Ragueneau, 202 rue Saint Honoré, 75001 Paris, France
Le Caminito, 48 rue du Dessous des Berges, 75013 Paris, France
Epicerie Musicale, 55bis quai de Valmy, 75010 Paris, France
Le petit Bretonneau, Le petit Bretonneau - à l'intérieur de l'Hôpital, 75018 Paris, France
Le Centenaire, 104 rue amelot, 75011 Paris, France
La Montagne Sans Geneviève, 13 Rue du Pot de Fer, 75005 Paris, France
Les Pères Populaires, 46 rue de Buzenval, 75020 Paris, France
Cafe de grenelle, 188 rue de Grenelle, 75007 Paris, France
Le relais de la victoire, 73 rue de la Victoire, 75009 Paris, France
Le Caminito, 48 rue du Dessous des Berges, 75013 Paris, France
Le petit Bretonneau, Le petit Bretonneau - à l'intérieur de l'Hôpital, 75018 Paris, France
La chaumière gourmande, Route de la Muette à Neuilly
Club hippique du Jardin d’Acclimatation, 75016 Paris, France
Le Chaumontois, 12 rue Armand Carrel, 75018 Paris, France
Caves populaires, 22 rue des Dames, 75017 Paris, France
Caprice café, 12 avenue Jean Moulin, 75014 Paris, France
Tamm Bara, 7 rue Clisson, 75013 Paris, France
L'anjou, 1 rue de Montholon, 75009 Paris, France
Café dans l'aerogare Air France Invalides, 2 rue Robert Esnault Pelterie, 75007 Paris, France
Chez Prune, 36 rue Beaurepaire, 75010 Paris, France
Au Vin Des Rues, 21 rue Boulard, 75014 Paris, France
bistrot les timbrés, 14 rue d'alleray, 75015 Paris, France
Café beauveau, 9 rue de Miromesnil, 75008 Paris, France
Le bal du pirate, 60 rue des bergers, 75015 Paris, France
Le Zazabar, 116 Rue de Ménilmontant, 75020 Paris, France
L'antre d'eux, 16 rue DE MEZIERES, 75006 Paris, France
l'orillon bar, 35 rue de l'orillon, 75011 Paris, France
zic zinc, 95 rue claude decaen, 75012 Paris, France
Les Pères Populaires, 46 rue de Buzenval, 75020 Paris, France
Epicerie Musicale, 55bis quai de Valmy, 75010 Paris, France
Le relais de la victoire, 73 rue de la Victoire, 75009 Paris, France
Le Centenaire, 104 rue amelot, 75011 Paris, France
Cafe de grenelle, 188 rue de Grenelle, 75007 Paris, France
Ragueneau, 202 rue Saint Honoré, 75001 Paris, France
Café Pistache, 9 rue des petits champs, 75001 Paris, France
La Cagnotte, 13 Rue Jean-Baptiste Dumay, 75020 Paris, France
le 1 cinq, 172 rue de vaugirard, 75015 Paris, France
Le Killy Jen, 28 bis boulevard Diderot, 75012 Paris, France
Café beauveau, 9 rue de Miromesnil, 75008 Paris, France
le 1 cinq, 172 rue de vaugirard, 75015 Paris, France
Les Artisans, 106 rue Lecourbe, 75015 Paris, France
Peperoni, 83 avenue de Wagram, 75001 Paris, France
le lutece, 380 rue de vaugirard, 75015 Paris, France
Brasiloja, 16 rue Ganneron, 75018 Paris, France
Rivolux, 16 rue de Rivoli, 75004 Paris, France
Chai 33, 33 Cour Saint Emilion, 75012 Paris, France
L'européen, 21 Bis Boulevard Diderot, 75012 Paris, France
NoMa, 39 rue Notre Dame de Nazareth, 75003 Paris, France
O'Paris, 1 Rue des Envierges, 75020 Paris, France
Café Clochette, 16 avenue Richerand, 75010 Paris, France
La cantoche de Paname, 40 Boulevard Beaumarchais, 75011 Paris, France
Le Saint René, 148 Boulevard de Charonne, 75020 Paris, France
La Liberté, 196 rue du faubourg saint-antoine, 75012 Paris, France
Chez Rutabaga, 16 rue des Petits Champs, 75002 Paris, France
Le BB (Bouchon des Batignolles), 2 rue Lemercier, 75017 Paris, France
La Brocante, 10 rue Rossini, 75009 Paris, France
Le Plomb du cantal, 3 rue Gaîté, 75014 Paris, France
Les caves populaires, 22 rue des Dames, 75017 Paris, France
Chez Luna, 108 rue de Ménilmontant, 75020 Paris, France
Le bar Fleuri, 1 rue du Plateau, 75019 Paris, France
Trois pièces cuisine, 101 rue des dames, 75017 Paris, France
Le Zinc, 61 avenue de la Motte Picquet, 75015 Paris, France
La cantine de Zoé, 136 rue du Faubourg poissonnière, 75010 Paris, France
Les Vendangeurs, 6/8 rue Stanislas, 75006 Paris, France
L'avant comptoir, 3 carrefour de l'Odéon, 75006 Paris, France
Le Brio, 216, rue Marcadet, 75018 Paris, France
Tamm Bara, 7 rue Clisson, 75013 Paris, France
Café dans l'aerogare Air France Invalides, 2 rue Robert Esnault Pelterie, 75007 Paris, France
bistrot les timbrés, 14 rue d'alleray, 75015 Paris, France
Caprice café, 12 avenue Jean Moulin, 75014 Paris, France
Caves populaires, 22 rue des Dames, 75017 Paris, France
Au Vin Des Rues, 21 rue Boulard, 75014 Paris, France
Chez Prune, 36 rue Beaurepaire, 75010 Paris, France
L'Inévitable, 22 rue Linné, 75005 Paris, France
L'anjou, 1 rue de Montholon, 75009 Paris, France
Botak cafe, 1 rue Paul albert, 75018 Paris, France
le chateau d'eau, 67 rue du Château d'eau, 75010 Paris, France
Bistrot Saint-Antoine, 58 rue du Fbg Saint-Antoine, 75012 Paris, France
Chez Oscar, 11/13 boulevard Beaumarchais, 75004 Paris, France
Le Fronton, 63 rue de Ponthieu, 75008 Paris, France
Le Piquet, 48 avenue de la Motte Picquet, 75015 Paris, France
Le Tournebride, 104 rue Mouffetard, 75005 Paris, France
L'avant comptoir, 3 carrefour de l'Odéon, 75006 Paris, France
le chateau d'eau, 67 rue du Château d'eau, 75010 Paris, France
Les Vendangeurs, 6/8 rue Stanislas, 75006 Paris, France
maison du vin, 52 rue des plantes, 75014 Paris, France
Coffee Chope, 344Vrue Vaugirard, 75015 Paris, France
L'entrepôt, 157 rue Bercy 75012 Paris, 75012 Paris, France
Le Tournebride, 104 rue Mouffetard, 75005 Paris, France
Le Fronton, 63 rue de Ponthieu, 75008 Paris, France
Le BB (Bouchon des Batignolles), 2 rue Lemercier, 75017 Paris, France
La cantine de Zoé, 136 rue du Faubourg poissonnière, 75010 Paris, France
Chez Rutabaga, 16 rue des Petits Champs, 75002 Paris, France
Les caves populaires, 22 rue des Dames, 75017 Paris, France
Le Plomb du cantal, 3 rue Gaîté, 75014 Paris, France
Trois pièces cuisine, 101 rue des dames, 75017 Paris, France
La Brocante, 10 rue Rossini, 75009 Paris, France
Le Zinc, 61 avenue de la Motte Picquet, 75015 Paris, France
Chez Luna, 108 rue de Ménilmontant, 75020 Paris, France
Le bar Fleuri, 1 rue du Plateau, 75019 Paris, France
La Liberté, 196 rue du faubourg saint-antoine, 75012 Paris, France
La cantoche de Paname, 40 Boulevard Beaumarchais, 75011 Paris, France
Le Saint René, 148 Boulevard de Charonne, 75020 Paris, France
Café Clochette, 16 avenue Richerand, 75010 Paris, France
L'européen, 21 Bis Boulevard Diderot, 75012 Paris, France
NoMa, 39 rue Notre Dame de Nazareth, 75003 Paris, France
le lutece, 380 rue de vaugirard, 75015 Paris, France
O'Paris, 1 Rue des Envierges, 75020 Paris, France
Rivolux, 16 rue de Rivoli, 75004 Paris, France
Brasiloja, 16 rue Ganneron, 75018 Paris, France
Institut des Cultures d'Islam, 19-23 rue Léon, 75018 Paris, France
Canopy Café associatif, 19 rue Pajol, 75018 Paris, France
Petits Freres des Pauvres, 47 rue de Batignolles, 75017 Paris, France
Le Lucernaire, 53 rue Notre-Dame des Champs, 75006 Paris, France
L'Angle, 28 rue de Ponthieu, 75008 Paris, France
Le Café d'avant, 35 rue Claude Bernard, 75005 Paris, France
Café Dupont, 198 rue de la Convention, 75015 Paris, France
Le Sévigné, 15 rue du Parc Royal, 75003 Paris, France
L'Entracte, place de l'opera, 75002 Paris, France
Panem, 18 rue de Crussol, 75011 Paris, France
Au pays de Vannes, 34 bis rue de Wattignies, 75012 Paris, France
l'Eléphant du nil, 125 Rue Saint-Antoine, 75004 Paris, France
L'âge d'or, 26 rue du Docteur Magnan, 75013 Paris, France
Le Comptoir, 354 bis rue Vaugirard, 75015 Paris, France
L'horizon, 93, rue de la Roquette, 75011 Paris, France
L'empreinte, 54, avenue Daumesnil, 75012 Paris, France
Café Victor, 10 boulevard Victor, 75015 Paris, France
Café Varenne, 36 rue de Varenne, 75007 Paris, France
Le Brigadier, 12 rue Blanche, 75009 Paris, France
Waikiki, 10 rue d"Ulm, 75005 Paris, France
Le Parc Vaugirard, 358 rue de Vaugirard, 75015 Paris, France
Pari's Café, 174 avenue de Clichy, 75017 Paris, France
Melting Pot, 3 rue de Lagny, 75020 Paris, France
le Zango, 58 rue Daguerre, 75014 Paris, France
Chez Miamophile, 6 rue Mélingue, 75019 Paris, France
Le café Monde et Médias, Place de la République, 75003 Paris, France
Café rallye tournelles, 11 Quai de la Tournelle, 75005 Paris, France
Brasserie le Morvan, 61 rue du château d'eau, 75010 Paris, France
Chez Miamophile, 6 rue Mélingue, 75019 Paris, France
La Caravane, Rue de la Fontaine au Roi, 75011 Paris, France
Panem, 18 rue de Crussol, 75011 Paris, France
Petits Freres des Pauvres, 47 rue de Batignolles, 75017 Paris, France
Café Dupont, 198 rue de la Convention, 75015 Paris, France
L'Angle, 28 rue de Ponthieu, 75008 Paris, France
Institut des Cultures d'Islam, 19-23 rue Léon, 75018 Paris, France
Canopy Café associatif, 19 rue Pajol, 75018 Paris, France
L'Entracte, place de l'opera, 75002 Paris, France
Le Sévigné, 15 rue du Parc Royal, 75003 Paris, France
Le Café d'avant, 35 rue Claude Bernard, 75005 Paris, France
Le Lucernaire, 53 rue Notre-Dame des Champs, 75006 Paris, France
Le Brigadier, 12 rue Blanche, 75009 Paris, France
L'âge d'or, 26 rue du Docteur Magnan, 75013 Paris, France
Café Victor, 10 boulevard Victor, 75015 Paris, France
L'empreinte, 54, avenue Daumesnil, 75012 Paris, France
L'horizon, 93, rue de la Roquette, 75011 Paris, France
Waikiki, 10 rue d"Ulm, 75005 Paris, France
Au pays de Vannes, 34 bis rue de Wattignies, 75012 Paris, France
Café Varenne, 36 rue de Varenne, 75007 Paris, France
l'Eléphant du nil, 125 Rue Saint-Antoine, 75004 Paris, France
Le Comptoir, 354 bis rue Vaugirard, 75015 Paris, France
Le Parc Vaugirard, 358 rue de Vaugirard, 75015 Paris, France
le Zango, 58 rue Daguerre, 75014 Paris, France
Melting Pot, 3 rue de Lagny, 75020 Paris, France
Pari's Café, 174 avenue de Clichy, 75017 Paris, France
L'entrepôt, 157 rue Bercy 75012 Paris, 75012 Paris, France

2
bonobo/examples/datasets/fablabs.py
View File

@ -96,8 +96,8 @@ graph = bonobo.Graph(
),
normalize,
filter_france,
bonobo.JsonWriter(path='fablabs.txt', ioformat='arg0'),
bonobo.Tee(display),
bonobo.JsonWriter(path='fablabs.txt'),
)
if __name__ == '__main__':

BIN
bonobo/examples/datasets/spam.tgz Normal file
View File

Binary file not shown.

4
bonobo/examples/files/csv_handlers.py
View File

@ -2,8 +2,8 @@ import bonobo
from bonobo.commands.run import get_default_services
graph = bonobo.Graph(
bonobo.CsvReader('datasets/coffeeshops.txt'),
print,
bonobo.CsvReader('datasets/coffeeshops.txt', headers=('item', )),
bonobo.PrettyPrinter(),
)
if __name__ == '__main__':

7
bonobo/examples/files/json_handlers.py
View File

@ -1,15 +1,16 @@
import bonobo
from bonobo import Bag
from bonobo.commands.run import get_default_services
def get_fields(row):
return row['fields']
def get_fields(**row):
return Bag(**row['fields'])
graph = bonobo.Graph(
bonobo.JsonReader('datasets/theaters.json'),
get_fields,
bonobo.PrettyPrint(title_keys=('eq_nom_equipement', )),
bonobo.PrettyPrinter(),
)
if __name__ == '__main__':

59
bonobo/examples/files/pickle_handlers.py Normal file
View File

@ -0,0 +1,59 @@
'''
This example shows how a different file system service can be injected
into a transformation (as compressing pickled objects often makes sense
anyways). The pickle itself contains a list of lists as follows:
```
[
['category', 'sms'],
['ham', 'Go until jurong point, crazy..'],
['ham', 'Ok lar... Joking wif u oni...'],
['spam', 'Free entry in 2 a wkly comp to win...'],
['ham', 'U dun say so early hor... U c already then say...'],
['ham', 'Nah I don't think he goes to usf, he lives around here though'],
['spam', 'FreeMsg Hey there darling it's been 3 week's now...'],
...
]
```
where the first column categorizes and sms as "ham" or "spam". The second
column contains the sms itself.
Data set taken from:
https://www.kaggle.com/uciml/sms-spam-collection-dataset/downloads/sms-spam-collection-dataset.zip
The transformation (1) reads the pickled data, (2) marks and shortens
messages categorized as spam, and (3) prints the output.
'''
import bonobo
from bonobo.commands.run import get_default_services
from fs.tarfs import TarFS
def cleanse_sms(**row):
if row['category'] == 'spam':
row['sms_clean'] = '**MARKED AS SPAM** ' + row['sms'][0:50] + (
'...' if len(row['sms']) > 50 else ''
)
else:
row['sms_clean'] = row['sms']
return row['sms_clean']
graph = bonobo.Graph(
# spam.pkl is within the gzipped tarball
bonobo.PickleReader('spam.pkl'),
cleanse_sms,
bonobo.PrettyPrinter(),
)
def get_services():
return {'fs': TarFS(bonobo.get_examples_path('datasets/spam.tgz'))}
if __name__ == '__main__':
bonobo.run(graph, services=get_default_services(__file__))

0
bonobo/examples/utils/__init__.py → bonobo/examples/nodes/__init__.py
View File

5
bonobo/examples/nodes/_services.py Normal file
View File

@ -0,0 +1,5 @@
from bonobo import get_examples_path, open_fs
def get_services():
return {'fs': open_fs(get_examples_path())}

0
bonobo/examples/utils/count.py → bonobo/examples/nodes/count.py
View File

24
bonobo/examples/nodes/filter.py Normal file
View File

@ -0,0 +1,24 @@
import bonobo
from bonobo import Filter
class OddOnlyFilter(Filter):
def filter(self, i):
return i % 2
@Filter
def multiples_of_three(i):
return not (i % 3)
graph = bonobo.Graph(
lambda: tuple(range(50)),
OddOnlyFilter(),
multiples_of_three,
print,
)
if __name__ == '__main__':
bonobo.run(graph)

19
bonobo/examples/nodes/slow.py Normal file
View File

@ -0,0 +1,19 @@
import bonobo
import time
from bonobo.constants import NOT_MODIFIED
def pause(*args, **kwargs):
time.sleep(0.1)
return NOT_MODIFIED
graph = bonobo.Graph(
lambda: tuple(range(20)),
pause,
print,
)
if __name__ == '__main__':
bonobo.run(graph)

23
bonobo/examples/tutorials/tut01e01.py Normal file
View File

@ -0,0 +1,23 @@
import bonobo
def extract():
yield 'foo'
yield 'bar'
yield 'baz'
def transform(x):
return x.upper()
def load(x):
print(x)
graph = bonobo.Graph(extract, transform, load)
graph.__doc__ = 'hello'
if __name__ == '__main__':
bonobo.run(graph)

14
bonobo/examples/tutorials/tut01e02.py Normal file
View File

@ -0,0 +1,14 @@
import bonobo
graph = bonobo.Graph(
[
'foo',
'bar',
'baz',
],
str.upper,
print,
)
if __name__ == '__main__':
bonobo.run(graph)

11
bonobo/examples/tutorials/tut02_01_read.py
View File

@ -1,11 +0,0 @@
import bonobo
graph = bonobo.Graph(
bonobo.FileReader('coffeeshops.txt'),
print,
)
if __name__ == '__main__':
bonobo.run(
graph, services={'fs': bonobo.open_examples_fs('datasets')}
)

17
bonobo/examples/tutorials/tut02_02_write.py
View File

@ -1,17 +0,0 @@
import bonobo
def split_one(line):
return line.split(', ', 1)
graph = bonobo.Graph(
bonobo.FileReader('coffeeshops.txt'),
split_one,
bonobo.JsonWriter('coffeeshops.json'),
)
if __name__ == '__main__':
bonobo.run(
graph, services={'fs': bonobo.open_examples_fs('datasets')}
)

14
bonobo/examples/tutorials/tut02e01_read.py Normal file
View File

@ -0,0 +1,14 @@
import bonobo
graph = bonobo.Graph(
bonobo.FileReader('coffeeshops.txt'),
print,
)
def get_services():
return {'fs': bonobo.open_examples_fs('datasets')}
if __name__ == '__main__':
bonobo.run(graph, services=get_services())

25
bonobo/examples/tutorials/tut02e02_write.py Normal file
View File

@ -0,0 +1,25 @@
import bonobo
def split_one(line):
return line.split(', ', 1)
graph = bonobo.Graph(
bonobo.FileReader('coffeeshops.txt'),
split_one,
bonobo.JsonWriter(
'coffeeshops.json', fs='fs.output', ioformat='arg0'
),
)
def get_services():
return {
'fs': bonobo.open_examples_fs('datasets'),
'fs.output': bonobo.open_fs(),
}
if __name__ == '__main__':
bonobo.run(graph, services=get_services())

18
bonobo/examples/tutorials/tut02_03_writeasmap.py → bonobo/examples/tutorials/tut02e03_writeasmap.py
View File

@ -1,4 +1,6 @@
import bonobo, json
import json
import bonobo
def split_one_to_map(line):
@ -18,10 +20,16 @@ class MyJsonWriter(bonobo.JsonWriter):
graph = bonobo.Graph(
bonobo.FileReader('coffeeshops.txt'),
split_one_to_map,
MyJsonWriter('coffeeshops.json'),
MyJsonWriter('coffeeshops.json', fs='fs.output', ioformat='arg0'),
)
def get_services():
return {
'fs': bonobo.open_examples_fs('datasets'),
'fs.output': bonobo.open_fs(),
}
if __name__ == '__main__':
bonobo.run(
graph, services={'fs': bonobo.open_examples_fs('datasets')}
)
bonobo.run(graph, services=get_services())

3
bonobo/examples/types/__main__.py Normal file
View File

@ -0,0 +1,3 @@
from bonobo.util.python import require
graph = require('strings').graph

29
bonobo/execution/base.py
View File

@ -9,6 +9,14 @@ from bonobo.util.errors import print_error
from bonobo.util.objects import Wrapper, get_name
@contextmanager
def recoverable(error_handler):
try:
yield
except Exception as exc: # pylint: disable=broad-except
error_handler(exc, traceback.format_exc())
@contextmanager
def unrecoverable(error_handler):
try:
@ -50,15 +58,21 @@ class LoopingExecutionContext(Wrapper):
# XXX enhancers
self._enhancers = get_enhancers(self.wrapped)
def __enter__(self):
self.start()
return self
def __exit__(self, exc_type=None, exc_val=None, exc_tb=None):
self.stop()
def start(self):
if self.started:
raise RuntimeError('Cannot start a node twice ({}).'.format(get_name(self)))
self._started = True
self._stack = ContextCurrifier(self.wrapped, *self._get_initial_context())
with unrecoverable(self.handle_error):
self._stack.setup(self)
self._stack = ContextCurrifier(self.wrapped, *self._get_initial_context())
self._stack.setup(self)
for enhancer in self._enhancers:
with unrecoverable(self.handle_error):
@ -81,10 +95,11 @@ class LoopingExecutionContext(Wrapper):
if self._stopped:
return
self._stopped = True
with unrecoverable(self.handle_error):
self._stack.teardown()
try:
if self._stack:
self._stack.teardown()
finally:
self._stopped = True
def handle_error(self, exc, trace):
return print_error(exc, trace, context=self.wrapped)

10
bonobo/execution/graph.py
View File

@ -21,16 +21,12 @@ class GraphExecutionContext:
def __init__(self, graph, plugins=None, services=None):
self.graph = graph
self.nodes = [NodeExecutionContext(node, parent=self) for node in self.graph.nodes]
self.nodes = [NodeExecutionContext(node, parent=self) for node in self.graph]
self.plugins = [PluginExecutionContext(plugin, parent=self) for plugin in plugins or ()]
self.services = Container(services) if services else Container()
for i, node_context in enumerate(self):
try:
node_context.outputs = [self[j].input for j in self.graph.outputs_of(i)]
except KeyError:
continue
node_context.outputs = [self[j].input for j in self.graph.outputs_of(i)]
node_context.input.on_begin = partial(node_context.send, BEGIN, _control=True)
node_context.input.on_end = partial(node_context.send, END, _control=True)
node_context.input.on_finalize = partial(node_context.stop)
@ -65,4 +61,4 @@ class GraphExecutionContext:
def stop(self):
# todo use strategy
for node in self.nodes:
node.stop()
node.stop()

16
bonobo/execution/node.py
View File

@ -3,15 +3,15 @@ from queue import Empty
from time import sleep
from bonobo.constants import INHERIT_INPUT, NOT_MODIFIED
from bonobo.core.inputs import Input
from bonobo.core.statistics import WithStatistics
from bonobo.errors import InactiveReadableError
from bonobo.errors import InactiveReadableError, UnrecoverableError
from bonobo.execution.base import LoopingExecutionContext
from bonobo.structs.bags import Bag
from bonobo.structs.inputs import Input
from bonobo.util.compat import deprecated_alias
from bonobo.util.errors import is_error
from bonobo.util.iterators import iter_if_not_sequence
from bonobo.util.objects import get_name
from bonobo.util.statistics import WithStatistics
class NodeExecutionContext(WithStatistics, LoopingExecutionContext):
@ -22,7 +22,7 @@ class NodeExecutionContext(WithStatistics, LoopingExecutionContext):
@property
def alive(self):
"""todo check if this is right, and where it is used"""
return self.input.alive and self._started and not self._stopped
return self._started and not self._stopped
@property
def alive_str(self):
@ -42,7 +42,7 @@ class NodeExecutionContext(WithStatistics, LoopingExecutionContext):
name, type_name = get_name(self), get_name(type(self))
return '<{}({}{}){}>'.format(type_name, self.alive_str, name, self.get_statistics_as_string(prefix=' '))
def write(self, *messages): # XXX write() ? ( node.write(...) )
def write(self, *messages):
"""
Push a message list to this context's input queue.
@ -54,7 +54,7 @@ class NodeExecutionContext(WithStatistics, LoopingExecutionContext):
# XXX deprecated alias
recv = deprecated_alias('recv', write)
def send(self, value, _control=False): # XXX self.send(....)
def send(self, value, _control=False):
"""
Sends a message to all of this context's outputs.
@ -93,6 +93,10 @@ class NodeExecutionContext(WithStatistics, LoopingExecutionContext):
except Empty:
sleep(self.PERIOD)
continue
except UnrecoverableError as exc:
self.handle_error(exc, traceback.format_exc())
self.input.shutdown()
break
except Exception as exc: # pylint: disable=broad-except
self.handle_error(exc, traceback.format_exc())

22
bonobo/execution/plugin.py
View File

@ -1,6 +1,4 @@
import traceback
from bonobo.execution.base import LoopingExecutionContext
from bonobo.execution.base import LoopingExecutionContext, recoverable
class PluginExecutionContext(LoopingExecutionContext):
@ -14,21 +12,15 @@ class PluginExecutionContext(LoopingExecutionContext):
def start(self):
super().start()
try:
with recoverable(self.handle_error):
self.wrapped.initialize()
except Exception as exc: # pylint: disable=broad-except
self.handle_error(exc, traceback.format_exc())
def shutdown(self):
try:
self.wrapped.finalize()
except Exception as exc: # pylint: disable=broad-except
self.handle_error(exc, traceback.format_exc())
finally:
self.alive = False
if self.started:
with recoverable(self.handle_error):
self.wrapped.finalize()
self.alive = False
def step(self):
try:
with recoverable(self.handle_error):
self.wrapped.run()
except Exception as exc: # pylint: disable=broad-except
self.handle_error(exc, traceback.format_exc())

140
bonobo/ext/console.py Normal file
View File

@ -0,0 +1,140 @@
import io
import sys
from contextlib import redirect_stdout
from colorama import Style, Fore, init
init(wrap=True)
from bonobo import settings
from bonobo.plugins import Plugin
from bonobo.util.term import CLEAR_EOL, MOVE_CURSOR_UP
class IOBuffer():
"""
The role of IOBuffer is to overcome the problem of multiple threads wanting to write to stdout at the same time. It
works a bit like a videogame: there are two buffers, one that is used to write, and one which is used to read from.
On each cycle, we swap the buffers, and the console plugin handle output of the one which is not anymore "active".
"""
def __init__(self):
self.current = io.StringIO()
self.write = self.current.write
def switch(self):
previous = self.current
self.current = io.StringIO()
self.write = self.current.write
try:
return previous.getvalue()
finally:
previous.close()
def flush(self):
self.current.flush()
class ConsoleOutputPlugin(Plugin):
"""
Outputs status information to the connected stdout. Can be a TTY, with or without support for colors/cursor
movements, or a non tty (pipe, file, ...). The features are adapted to terminal capabilities.
On Windows, we'll play a bit differently because we don't know how to manipulate cursor position. We'll only
display stats at the very end, and there won't be this "buffering" logic we need to display both stats and stdout.
.. attribute:: prefix
String prefix of output lines.
"""
def initialize(self):
self.prefix = ''
self.counter = 0
self._append_cache = ''
self.isatty = sys.stdout.isatty()
self.iswindows = (sys.platform == 'win32')
self._stdout = sys.stdout
self.stdout = IOBuffer()
self.redirect_stdout = redirect_stdout(self._stdout if self.iswindows else self.stdout)
self.redirect_stdout.__enter__()
def run(self):
if self.isatty and not self.iswindows:
self._write(self.context.parent, rewind=True)
else:
pass # not a tty, or windows, so we'll ignore stats output
def finalize(self):
self._write(self.context.parent, rewind=False)
self.redirect_stdout.__exit__(None, None, None)
def write(self, context, prefix='', rewind=True, append=None):
t_cnt = len(context)
if not self.iswindows:
buffered = self.stdout.switch()
for line in buffered.split('\n')[:-1]:
print(line + CLEAR_EOL, file=sys.stderr)
alive_color = Style.BRIGHT
dead_color = (Style.BRIGHT + Fore.BLACK) if self.iswindows else Fore.BLACK
for i in context.graph.topologically_sorted_indexes:
node = context[i]
name_suffix = '({})'.format(i) if settings.DEBUG.get() else ''
if node.alive:
_line = ''.join(
(
' ', alive_color, '+', Style.RESET_ALL, ' ', node.name, name_suffix, ' ',
node.get_statistics_as_string(), Style.RESET_ALL, ' ',
)
)
else:
_line = ''.join(
(
' ', dead_color, '-', ' ', node.name, name_suffix, ' ', node.get_statistics_as_string(),
Style.RESET_ALL, ' ',
)
)
print(prefix + _line + '\033[0K', file=sys.stderr)
if append:
# todo handle multiline
print(
''.join(
(
' `-> ', ' '.join('{}{}{}: {}'.format(Style.BRIGHT, k, Style.RESET_ALL, v)
for k, v in append), CLEAR_EOL
)
),
file=sys.stderr
)
t_cnt += 1
if rewind:
print(CLEAR_EOL, file=sys.stderr)
print(MOVE_CURSOR_UP(t_cnt + 2), file=sys.stderr)
def _write(self, graph_context, rewind):
if settings.PROFILE.get():
if self.counter % 10 and self._append_cache:
append = self._append_cache
else:
self._append_cache = append = (
('Memory', '{0:.2f} Mb'.format(memory_usage())),
# ('Total time', '{0} s'.format(execution_time(harness))),
)
else:
append = ()
self.write(graph_context, prefix=self.prefix, append=append, rewind=rewind)
self.counter += 1
def memory_usage():
import os, psutil
process = psutil.Process(os.getpid())
return process.memory_info()[0] / float(2**20)

5
bonobo/ext/console/__init__.py
View File

@ -1,5 +0,0 @@
from .plugin import ConsoleOutputPlugin
__all__ = [
'ConsoleOutputPlugin',
]

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

@ -1,108 +0,0 @@
# -*- coding: utf-8 -*-
#
# Copyright 2012-2017 Romain Dorgueil
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import functools
import sys
from colorama import Fore, Style
from bonobo.plugins import Plugin
from bonobo.util.term import CLEAR_EOL, MOVE_CURSOR_UP
@functools.lru_cache(1)
def memory_usage():
import os, psutil
process = psutil.Process(os.getpid())
return process.get_memory_info()[0] / float(2**20)
# @lru_cache(64)
# def execution_time(harness):
# return datetime.datetime.now() - harness._started_at
class ConsoleOutputPlugin(Plugin):
"""
Outputs status information to the connected stdout. Can be a TTY, with or without support for colors/cursor
movements, or a non tty (pipe, file, ...). The features are adapted to terminal capabilities.
.. attribute:: prefix
String prefix of output lines.
"""
def initialize(self):
self.prefix = ''
def _write(self, graph_context, rewind):
profile, debug = False, False
if profile:
append = (
('Memory', '{0:.2f} Mb'.format(memory_usage())),
# ('Total time', '{0} s'.format(execution_time(harness))),
)
else:
append = ()
self.write(graph_context, prefix=self.prefix, append=append, debug=debug, profile=profile, rewind=rewind)
def run(self):
if sys.stdout.isatty():
self._write(self.context.parent, rewind=True)
else:
pass # not a tty
def finalize(self):
self._write(self.context.parent, rewind=False)
@staticmethod
def write(context, prefix='', rewind=True, append=None, debug=False, profile=False):
t_cnt = len(context)
for i, component in enumerate(context):
if component.alive:
_line = ''.join(
(
Fore.BLACK, '({})'.format(i + 1), Style.RESET_ALL, ' ', Style.BRIGHT, '+', Style.RESET_ALL, ' ',
component.name, ' ', component.get_statistics_as_string(debug=debug,
profile=profile), Style.RESET_ALL, ' ',
)
)
else:
_line = ''.join(
(
Fore.BLACK, '({})'.format(i + 1), ' - ', component.name, ' ',
component.get_statistics_as_string(debug=debug, profile=profile), Style.RESET_ALL, ' ',
)
)
print(prefix + _line + '\033[0K')
if append:
# todo handle multiline
print(
''.join(
(
' `-> ', ' '.join('{}{}{}: {}'.format(Style.BRIGHT, k, Style.RESET_ALL, v)
for k, v in append), CLEAR_EOL
)
)
)
t_cnt += 1
if rewind:
print(CLEAR_EOL)
print(MOVE_CURSOR_UP(t_cnt + 2))

0
bonobo/ext/edgy/project/__init__.py
View File

22
bonobo/ext/edgy/project/feature.py
View File

@ -1,22 +0,0 @@
try:
import edgy.project
except ImportError as e:
import logging
logging.exception('You must install edgy.project to use this.')
import os
from edgy.project.events import subscribe
from edgy.project.feature import Feature, SUPPORT_PRIORITY
class BonoboFeature(Feature):
requires = {'python'}
@subscribe('edgy.project.on_start', priority=SUPPORT_PRIORITY)
def on_start(self, event):
package_path = event.setup['name'].replace('.', os.sep)
for file in ('example_graph'):
self.render_file(os.path.join(package_path, file + '.py'), os.path.join('tornado', file + '.py.j2'))

21
bonobo/ext/jupyter/js/dist/index.js vendored
View File

@ -42,7 +42,7 @@ define(["jupyter-js-widgets"], function(__WEBPACK_EXTERNAL_MODULE_2__) { return
/************************************************************************/
/******/ ([
/* 0 */
/***/ function(module, exports, __webpack_require__) {
/***/ (function(module, exports, __webpack_require__) {
// Entry point for the unpkg bundle containing custom model definitions.
//
@ -55,14 +55,13 @@ define(["jupyter-js-widgets"], function(__WEBPACK_EXTERNAL_MODULE_2__) { return
module.exports['version'] = __webpack_require__(4).version;
/***/ },
/***/ }),
/* 1 */
/***/ function(module, exports, __webpack_require__) {
/***/ (function(module, exports, __webpack_require__) {
var widgets = __webpack_require__(2);
var _ = __webpack_require__(3);
// Custom Model. Custom widgets models must at least provide default values
// for model attributes, including `_model_name`, `_view_name`, `_model_module`
// and `_view_module` when different from the base class.
@ -102,15 +101,15 @@ define(["jupyter-js-widgets"], function(__WEBPACK_EXTERNAL_MODULE_2__) { return
};
/***/ },
/***/ }),
/* 2 */
/***/ function(module, exports) {
/***/ (function(module, exports) {
module.exports = __WEBPACK_EXTERNAL_MODULE_2__;
/***/ },
/***/ }),
/* 3 */
/***/ function(module, exports, __webpack_require__) {
/***/ (function(module, exports, __webpack_require__) {
var __WEBPACK_AMD_DEFINE_ARRAY__, __WEBPACK_AMD_DEFINE_RESULT__;// Underscore.js 1.8.3
// http://underscorejs.org
@ -1662,9 +1661,9 @@ define(["jupyter-js-widgets"], function(__WEBPACK_EXTERNAL_MODULE_2__) { return
}.call(this));
/***/ },
/***/ }),
/* 4 */
/***/ function(module, exports) {
/***/ (function(module, exports) {
module.exports = {
"name": "bonobo-jupyter",
@ -1696,6 +1695,6 @@ define(["jupyter-js-widgets"], function(__WEBPACK_EXTERNAL_MODULE_2__) { return
}
};
/***/ }
/***/ })
/******/ ])});;
//# sourceMappingURL=index.js.map

2
bonobo/ext/jupyter/js/dist/index.js.map vendored
View File

File diff suppressed because one or more lines are too long

1
bonobo/ext/jupyter/js/src/bonobo.js
View File

@ -1,7 +1,6 @@
var widgets = require('jupyter-js-widgets');
var _ = require('underscore');
// Custom Model. Custom widgets models must at least provide default values
// for model attributes, including `_model_name`, `_view_name`, `_model_module`
// and `_view_module` when different from the base class.

1441
bonobo/ext/jupyter/js/yarn.lock Normal file
View File

File diff suppressed because it is too large Load Diff

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

@ -1,11 +1,11 @@
import logging
from bonobo.ext.jupyter.widget import BonoboWidget
from bonobo.plugins import Plugin
try:
import IPython.core.display
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 '
@ -19,6 +19,8 @@ class JupyterOutputPlugin(Plugin):
IPython.core.display.display(self.widget)
def run(self):
self.widget.value = [repr(node) for node in self.context.parent.nodes]
self.widget.value = [
str(self.context.parent[i]) for i in self.context.parent.graph.topologically_sorted_indexes
]
finalize = run

4
bonobo/ext/jupyter/static/extension.js
View File

@ -42,7 +42,7 @@ define(function() { return /******/ (function(modules) { // webpackBootstrap
/************************************************************************/
/******/ ([
/* 0 */
/***/ function(module, exports) {
/***/ (function(module, exports) {
// This file contains the javascript that is run when the notebook is loaded.
// It contains some requirejs configuration and the `load_ipython_extension`
@ -66,5 +66,5 @@ define(function() { return /******/ (function(modules) { // webpackBootstrap
};
/***/ }
/***/ })
/******/ ])});;

21
bonobo/ext/jupyter/static/index.js
View File

@ -42,7 +42,7 @@ define(["jupyter-js-widgets"], function(__WEBPACK_EXTERNAL_MODULE_2__) { return
/************************************************************************/
/******/ ([
/* 0 */
/***/ function(module, exports, __webpack_require__) {
/***/ (function(module, exports, __webpack_require__) {
// Entry point for the notebook bundle containing custom model definitions.
//
@ -58,14 +58,13 @@ define(["jupyter-js-widgets"], function(__WEBPACK_EXTERNAL_MODULE_2__) { return
module.exports['version'] = __webpack_require__(4).version;
/***/ },
/***/ }),
/* 1 */
/***/ function(module, exports, __webpack_require__) {
/***/ (function(module, exports, __webpack_require__) {
var widgets = __webpack_require__(2);
var _ = __webpack_require__(3);
// Custom Model. Custom widgets models must at least provide default values
// for model attributes, including `_model_name`, `_view_name`, `_model_module`
// and `_view_module` when different from the base class.
@ -105,15 +104,15 @@ define(["jupyter-js-widgets"], function(__WEBPACK_EXTERNAL_MODULE_2__) { return
};
/***/ },
/***/ }),
/* 2 */
/***/ function(module, exports) {
/***/ (function(module, exports) {
module.exports = __WEBPACK_EXTERNAL_MODULE_2__;
/***/ },
/***/ }),
/* 3 */
/***/ function(module, exports, __webpack_require__) {
/***/ (function(module, exports, __webpack_require__) {
var __WEBPACK_AMD_DEFINE_ARRAY__, __WEBPACK_AMD_DEFINE_RESULT__;// Underscore.js 1.8.3
// http://underscorejs.org
@ -1665,9 +1664,9 @@ define(["jupyter-js-widgets"], function(__WEBPACK_EXTERNAL_MODULE_2__) { return
}.call(this));
/***/ },
/***/ }),
/* 4 */
/***/ function(module, exports) {
/***/ (function(module, exports) {
module.exports = {
"name": "bonobo-jupyter",
@ -1699,6 +1698,6 @@ define(["jupyter-js-widgets"], function(__WEBPACK_EXTERNAL_MODULE_2__) { return
}
};
/***/ }
/***/ })
/******/ ])});;
//# sourceMappingURL=index.js.map

2
bonobo/ext/jupyter/static/index.js.map
View File

File diff suppressed because one or more lines are too long

2
bonobo/ext/jupyter/widget.py
View File

@ -2,7 +2,7 @@ import ipywidgets as widgets
from traitlets import List, Unicode
@widgets.register('bonobo-widget.Bonobo')
@widgets.register('bonobo-widget.bonobo')
class BonoboWidget(widgets.DOMWidget):
_view_name = Unicode('BonoboView').tag(sync=True)
_model_name = Unicode('BonoboModel').tag(sync=True)

14
bonobo/ext/opendatasoft.py
View File

@ -3,9 +3,8 @@ from urllib.parse import urlencode
import requests # todo: make this a service so we can substitute it ?
from bonobo.config import Option
from bonobo.config.processors import ContextProcessor, contextual
from bonobo.config.processors import ContextProcessor
from bonobo.config.configurables import Configurable
from bonobo.util.compat import deprecated
from bonobo.util.objects import ValueHolder
@ -14,13 +13,13 @@ def path_str(path):
class OpenDataSoftAPI(Configurable):
dataset = Option(str, required=True)
dataset = Option(str, positional=True)
endpoint = Option(str, default='{scheme}://{netloc}{path}')
scheme = Option(str, default='https')
netloc = Option(str, default='data.opendatasoft.com')
path = Option(path_str, default='/api/records/1.0/search/')
rows = Option(int, default=500)
limit = Option(int, default=None)
limit = Option(int, required=False)
timezone = Option(str, default='Europe/Paris')
kwargs = Option(dict, default=dict)
@ -47,12 +46,7 @@ class OpenDataSoftAPI(Configurable):
for row in records:
yield {**row.get('fields', {}), 'geometry': row.get('geometry', {})}
start.value += self.rows
@deprecated
def from_opendatasoft_api(dataset, **kwargs):
return OpenDataSoftAPI(dataset=dataset, **kwargs)
start += self.rows
__all__ = [

0
bonobo/ext/pandas.py
View File

99
bonobo/io/file.py
View File

@ -1,99 +0,0 @@
from bonobo.config import Option, Service
from bonobo.config.configurables import Configurable
from bonobo.config.processors import ContextProcessor, contextual
from bonobo.util.objects import ValueHolder
__all__ = [
'FileReader',
'FileWriter',
]
class FileHandler(Configurable):
"""Abstract component factory for file-related components.
Args:
path (str): which path to use within the provided filesystem.
eol (str): which character to use to separate lines.
mode (str): which mode to use when opening the file.
fs (str): service name to use for filesystem.
"""
path = Option(str, required=True, positional=True) # type: str
eol = Option(str, default='\n') # type: str
mode = Option(str) # type: str
fs = Service('fs') # type: str
@ContextProcessor
def file(self, context, fs):
with self.open(fs) as file:
yield file
def open(self, fs):
return fs.open(self.path, self.mode)
class Reader(FileHandler):
"""Abstract component factory for readers.
"""
def __call__(self, *args):
yield from self.read(*args)
def read(self, *args):
raise NotImplementedError('Abstract.')
class Writer(FileHandler):
"""Abstract component factory for writers.
"""
def __call__(self, *args):
return self.write(*args)
def write(self, *args):
raise NotImplementedError('Abstract.')
class FileReader(Reader):
"""Component factory for file-like readers.
On its own, it can be used to read a file and yield one row per line, trimming the "eol" character at the end if
present. Extending it is usually the right way to create more specific file readers (like json, csv, etc.)
"""
mode = Option(str, default='r')
def read(self, fs, file):
"""
Write a row on the next line of given file.
Prefix is used for newlines.
"""
for line in file:
yield line.rstrip(self.eol)
class FileWriter(Writer):
"""Component factory for file or file-like writers.
On its own, it can be used to write in a file one line per row that comes into this component. Extending it is
usually the right way to create more specific file writers (like json, csv, etc.)
"""
mode = Option(str, default='w+')
@ContextProcessor
def lineno(self, context, fs, file):
lineno = ValueHolder(0, type=int)
yield lineno
def write(self, fs, file, lineno, row):
"""
Write a row on the next line of opened file in context.
"""
self._write_line(file, (self.eol if lineno.value else '') + row)
lineno.value += 1
def _write_line(self, file, line):
return file.write(line)

38
bonobo/io/json.py
View File

@ -1,38 +0,0 @@
import json
from bonobo.config.processors import ContextProcessor, contextual
from .file import FileWriter, FileReader
__all__ = [
'JsonWriter',
]
class JsonHandler():
eol = ',\n'
prefix, suffix = '[', ']'
class JsonReader(JsonHandler, FileReader):
loader = staticmethod(json.load)
def read(self, fs, file):
for line in self.loader(file):
yield line
class JsonWriter(JsonHandler, FileWriter):
@ContextProcessor
def envelope(self, context, fs, file, lineno):
file.write(self.prefix)
yield
file.write(self.suffix)
def write(self, fs, file, lineno, row):
"""
Write a json row on the next line of file pointed by ctx.file.
:param ctx:
:param row:
"""
return super().write(fs, file, lineno, json.dumps(row))

86
bonobo/logging.py Normal file
View File

@ -0,0 +1,86 @@
import logging
import sys
import textwrap
from logging import CRITICAL, DEBUG, ERROR, INFO, WARNING
from colorama import Fore, Style
from bonobo import settings
from bonobo.util.term import CLEAR_EOL
iswindows = (sys.platform == 'win32')
def get_format():
yield '{b}[%(fg)s%(levelname)s{b}][{w}'
yield '{b}][{w}'.join(('%(spent)04d', '%(name)s'))
yield '{b}]'
yield ' %(fg)s%(message)s{r}'
if not iswindows:
yield CLEAR_EOL
colors = {
'b': '' if iswindows else Fore.BLACK,
'w': '' if iswindows else Fore.LIGHTBLACK_EX,
'r': '' if iswindows else Style.RESET_ALL,
}
format = (''.join(get_format())).format(**colors)
class Filter(logging.Filter):
def filter(self, record):
record.spent = record.relativeCreated // 1000
if iswindows:
record.fg = ''
elif record.levelname == 'DEBG':
record.fg = Fore.LIGHTBLACK_EX
elif record.levelname == 'INFO':
record.fg = Fore.LIGHTWHITE_EX
elif record.levelname == 'WARN':
record.fg = Fore.LIGHTYELLOW_EX
elif record.levelname == 'ERR ':
record.fg = Fore.LIGHTRED_EX
elif record.levelname == 'CRIT':
record.fg = Fore.RED
else:
record.fg = Fore.LIGHTWHITE_EX
return True
class Formatter(logging.Formatter):
def formatException(self, ei):
tb = super().formatException(ei)
if iswindows:
return textwrap.indent(tb, ' | ')
else:
return textwrap.indent(tb, Fore.BLACK + ' | ' + Fore.WHITE)
def setup(level):
logging.addLevelName(DEBUG, 'DEBG')
logging.addLevelName(INFO, 'INFO')
logging.addLevelName(WARNING, 'WARN')
logging.addLevelName(ERROR, 'ERR ')
logging.addLevelName(CRITICAL, 'CRIT')
handler = logging.StreamHandler(sys.stderr)
handler.setFormatter(Formatter(format))
handler.addFilter(Filter())
root = logging.getLogger()
root.addHandler(handler)
root.setLevel(level)
def set_level(level):
logging.getLogger().setLevel(level)
def get_logger(name='bonobo'):
return logging.getLogger(name)
# Compatibility with python logging
getLogger = get_logger
# Setup formating and level.
setup(level=settings.LOGGING_LEVEL.get())

8
bonobo/nodes/__init__.py Normal file
View File

@ -0,0 +1,8 @@
from bonobo.nodes.basics import *
from bonobo.nodes.basics import __all__ as _all_basics
from bonobo.nodes.filter import Filter
from bonobo.nodes.io import *
from bonobo.nodes.io import __all__ as _all_io
from bonobo.nodes.throttle import RateLimited
__all__ = _all_basics + _all_io + ['Filter', 'RateLimited']

112
bonobo/nodes/basics.py Normal file
View File

@ -0,0 +1,112 @@
import functools
import itertools
from bonobo import settings
from bonobo.config import Configurable, Option
from bonobo.config.processors import ContextProcessor
from bonobo.constants import NOT_MODIFIED
from bonobo.structs.bags import Bag
from bonobo.util.objects import ValueHolder
from bonobo.util.term import CLEAR_EOL
__all__ = [
'Limit',
'PrettyPrinter',
'Tee',
'arg0_to_kwargs',
'count',
'identity',
'kwargs_to_arg0',
'noop',
]
def identity(x):
return x
class Limit(Configurable):
"""
Creates a Limit() node, that will only let go through the first n rows (defined by the `limit` option), unmodified.
.. attribute:: limit
Number of rows to let go through.
"""
limit = Option(positional=True, default=10)
@ContextProcessor
def counter(self, context):
yield ValueHolder(0)
def call(self, counter, *args, **kwargs):
counter += 1
if counter <= self.limit:
yield NOT_MODIFIED
def Tee(f):
from bonobo.constants import NOT_MODIFIED
@functools.wraps(f)
def wrapped(*args, **kwargs):
nonlocal f
f(*args, **kwargs)
return NOT_MODIFIED
return wrapped
def count(counter, *args, **kwargs):
counter += 1
@ContextProcessor.decorate(count)
def _count_counter(self, context):
counter = ValueHolder(0)
yield counter
context.send(Bag(counter._value))
class PrettyPrinter(Configurable):
def call(self, *args, **kwargs):
formater = self._format_quiet if settings.QUIET.get() else self._format_console
for i, (item, value) in enumerate(itertools.chain(enumerate(args), kwargs.items())):
print(formater(i, item, value))
def _format_quiet(self, i, item, value):
return ' '.join(((' ' if i else '-'), str(item), ':', str(value).strip()))
def _format_console(self, i, item, value):
return ' '.join(
((' ' if i else '•'), str(item), '=', str(value).strip().replace('\n', '\n' + CLEAR_EOL), CLEAR_EOL)
)
def noop(*args, **kwargs): # pylint: disable=unused-argument
from bonobo.constants import NOT_MODIFIED
return NOT_MODIFIED
def arg0_to_kwargs(row):
"""
Transform items in a stream from "arg0" format (each call only has one positional argument, which is a dict-like
object) to "kwargs" format (each call only has keyword arguments that represent a row).
:param row:
:return: bonobo.Bag
"""
return Bag(**row)
def kwargs_to_arg0(**row):
"""
Transform items in a stream from "kwargs" format (each call only has keyword arguments that represent a row) to
"arg0" format (each call only has one positional argument, which is a dict-like object) .
:param **row:
:return: bonobo.Bag
"""
return Bag(row)

26
bonobo/nodes/filter.py Normal file
View File

@ -0,0 +1,26 @@
from bonobo.constants import NOT_MODIFIED
from bonobo.config import Configurable, Method
class Filter(Configurable):
"""Filter out hashes from the stream depending on the :attr:`filter` callable return value, when called with the
current hash as parameter.
Can be used as a decorator on a filter callable.
.. attribute:: filter
A callable used to filter lines.
If the callable returns a true-ish value, the input will be passed unmodified to the next items.
Otherwise, it'll be burnt.
"""
filter = Method()
def call(self, *args, **kwargs):
if self.filter(*args, **kwargs):
return NOT_MODIFIED

3
bonobo/io/__init__.py → bonobo/nodes/io/__init__.py
View File

@ -3,6 +3,7 @@
from .file import FileReader, FileWriter
from .json import JsonReader, JsonWriter
from .csv import CsvReader, CsvWriter
from .pickle import PickleReader, PickleWriter
__all__ = [
'CsvReader',
@ -11,4 +12,6 @@ __all__ = [
'FileWriter',
'JsonReader',
'JsonWriter',
'PickleReader',
'PickleWriter',
]

83
bonobo/nodes/io/base.py Normal file
View File

@ -0,0 +1,83 @@
from bonobo import settings
from bonobo.config import Configurable, ContextProcessor, Option, Service
from bonobo.errors import UnrecoverableValueError, UnrecoverableNotImplementedError
from bonobo.structs.bags import Bag
class IOFormatEnabled(Configurable):
ioformat = Option(default=settings.IOFORMAT.get)
def get_input(self, *args, **kwargs):
if self.ioformat == settings.IOFORMAT_ARG0:
if len(args) != 1 or len(kwargs):
raise UnrecoverableValueError(
'Wrong input formating: IOFORMAT=ARG0 implies one arg and no kwargs, got args={!r} and kwargs={!r}.'.
format(args, kwargs)
)
return args[0]
if self.ioformat == settings.IOFORMAT_KWARGS:
if len(args) or not len(kwargs):
raise UnrecoverableValueError(
'Wrong input formating: IOFORMAT=KWARGS ioformat implies no arg, got args={!r} and kwargs={!r}.'.
format(args, kwargs)
)
return kwargs
raise UnrecoverableNotImplementedError('Unsupported format.')
def get_output(self, row):
if self.ioformat == settings.IOFORMAT_ARG0:
return row
if self.ioformat == settings.IOFORMAT_KWARGS:
return Bag(**row)
raise UnrecoverableNotImplementedError('Unsupported format.')
class FileHandler(Configurable):
"""Abstract component factory for file-related components.
Args:
path (str): which path to use within the provided filesystem.
eol (str): which character to use to separate lines.
mode (str): which mode to use when opening the file.
fs (str): service name to use for filesystem.
"""
path = Option(str, required=True, positional=True) # type: str
eol = Option(str, default='\n') # type: str
mode = Option(str) # type: str
encoding = Option(str, default='utf-8') # type: str
fs = Service('fs') # type: str
@ContextProcessor
def file(self, context, fs):
with self.open(fs) as file:
yield file
def open(self, fs):
return fs.open(self.path, self.mode, encoding=self.encoding)
class Reader:
"""Abstract component factory for readers.
"""
def __call__(self, *args, **kwargs):
yield from self.read(*args, **kwargs)
def read(self, *args, **kwargs):
raise NotImplementedError('Abstract.')
class Writer:
"""Abstract component factory for writers.
"""
def __call__(self, *args, **kwargs):
return self.write(*args, **kwargs)
def write(self, *args, **kwargs):
raise NotImplementedError('Abstract.')

36
bonobo/io/csv.py → bonobo/nodes/io/csv.py
View File

@ -1,9 +1,11 @@
import csv
from bonobo.config import Option
from bonobo.config.processors import ContextProcessor, contextual
from bonobo.config.processors import ContextProcessor
from bonobo.constants import NOT_MODIFIED
from bonobo.nodes.io.file import FileReader, FileWriter
from bonobo.nodes.io.base import FileHandler, IOFormatEnabled
from bonobo.util.objects import ValueHolder
from .file import FileHandler, FileReader, FileWriter
class CsvHandler(FileHandler):
@ -24,10 +26,10 @@ class CsvHandler(FileHandler):
"""
delimiter = Option(str, default=';')
quotechar = Option(str, default='"')
headers = Option(tuple)
headers = Option(tuple, required=False)
class CsvReader(CsvHandler, FileReader):
class CsvReader(IOFormatEnabled, FileReader, CsvHandler):
"""
Reads a CSV and yield the values as dicts.
@ -45,8 +47,12 @@ class CsvReader(CsvHandler, FileReader):
def read(self, fs, file, headers):
reader = csv.reader(file, delimiter=self.delimiter, quotechar=self.quotechar)
headers.value = headers.value or next(reader)
field_count = len(headers.value)
if not headers.get():
headers.set(next(reader))
_headers = headers.get()
field_count = len(headers)
if self.skip and self.skip > 0:
for _ in range(0, self.skip):
@ -56,19 +62,21 @@ class CsvReader(CsvHandler, FileReader):
if len(row) != field_count:
raise ValueError('Got a line with %d fields, expecting %d.' % (len(row), field_count, ))
yield dict(zip(headers.value, row))
yield self.get_output(dict(zip(_headers, row)))
class CsvWriter(CsvHandler, FileWriter):
class CsvWriter(IOFormatEnabled, FileWriter, CsvHandler):
@ContextProcessor
def writer(self, context, fs, file, lineno):
writer = csv.writer(file, delimiter=self.delimiter, quotechar=self.quotechar, lineterminator=self.eol)
headers = ValueHolder(list(self.headers) if self.headers else None)
yield writer, headers
def write(self, fs, file, lineno, writer, headers, row):
if not lineno.value:
headers.value = headers.value or row.keys()
writer.writerow(headers.value)
writer.writerow(row[header] for header in headers.value)
lineno.value += 1
def write(self, fs, file, lineno, writer, headers, *args, **kwargs):
row = self.get_input(*args, **kwargs)
if not lineno:
headers.set(headers.value or row.keys())
writer.writerow(headers.get())
writer.writerow(row[header] for header in headers.get())
lineno += 1
return NOT_MODIFIED

49
bonobo/nodes/io/file.py Normal file
View File

@ -0,0 +1,49 @@
from bonobo.config import Option
from bonobo.config.processors import ContextProcessor
from bonobo.constants import NOT_MODIFIED
from bonobo.nodes.io.base import FileHandler, Reader, Writer
from bonobo.util.objects import ValueHolder
class FileReader(Reader, FileHandler):
"""Component factory for file-like readers.
On its own, it can be used to read a file and yield one row per line, trimming the "eol" character at the end if
present. Extending it is usually the right way to create more specific file readers (like json, csv, etc.)
"""
mode = Option(str, default='r')
def read(self, fs, file):
"""
Write a row on the next line of given file.
Prefix is used for newlines.
"""
for line in file:
yield line.rstrip(self.eol)
class FileWriter(Writer, FileHandler):
"""Component factory for file or file-like writers.
On its own, it can be used to write in a file one line per row that comes into this component. Extending it is
usually the right way to create more specific file writers (like json, csv, etc.)
"""
mode = Option(str, default='w+')
@ContextProcessor
def lineno(self, context, fs, file):
lineno = ValueHolder(0)
yield lineno
def write(self, fs, file, lineno, line):
"""
Write a row on the next line of opened file in context.
"""
self._write_line(file, (self.eol if lineno.value else '') + line)
lineno += 1
return NOT_MODIFIED
def _write_line(self, file, line):
return file.write(line)

39
bonobo/nodes/io/json.py Normal file
View File

@ -0,0 +1,39 @@
import json
from bonobo.config.processors import ContextProcessor
from bonobo.constants import NOT_MODIFIED
from bonobo.nodes.io.base import FileHandler, IOFormatEnabled
from bonobo.nodes.io.file import FileReader, FileWriter
class JsonHandler(FileHandler):
eol = ',\n'
prefix, suffix = '[', ']'
class JsonReader(IOFormatEnabled, FileReader, JsonHandler):
loader = staticmethod(json.load)
def read(self, fs, file):
for line in self.loader(file):
yield self.get_output(line)
class JsonWriter(IOFormatEnabled, FileWriter, JsonHandler):
@ContextProcessor
def envelope(self, context, fs, file, lineno):
file.write(self.prefix)
yield
file.write(self.suffix)
def write(self, fs, file, lineno, *args, **kwargs):
"""
Write a json row on the next line of file pointed by ctx.file.
:param ctx:
:param row:
"""
row = self.get_input(*args, **kwargs)
self._write_line(file, (self.eol if lineno.value else '') + json.dumps(row))
lineno += 1
return NOT_MODIFIED

69
bonobo/nodes/io/pickle.py Normal file
View File

@ -0,0 +1,69 @@
import pickle
from bonobo.config import Option
from bonobo.config.processors import ContextProcessor
from bonobo.constants import NOT_MODIFIED
from bonobo.nodes.io.base import FileHandler, IOFormatEnabled
from bonobo.nodes.io.file import FileReader, FileWriter
from bonobo.util.objects import ValueHolder
class PickleHandler(FileHandler):
"""
.. attribute:: item_names
The names of the items in the pickle, if it is not defined in the first item of the pickle.
"""
item_names = Option(tuple, required=False)
class PickleReader(IOFormatEnabled, FileReader, PickleHandler):
"""
Reads a Python pickle object and yields the items in dicts.
"""
mode = Option(str, default='rb')
@ContextProcessor
def pickle_headers(self, context, fs, file):
yield ValueHolder(self.item_names)
def read(self, fs, file, pickle_headers):
data = pickle.load(file)
# if the data is not iterable, then wrap the object in a list so it may be iterated
if isinstance(data, dict):
is_dict = True
iterator = iter(data.items())
else:
is_dict = False
try:
iterator = iter(data)
except TypeError:
iterator = iter([data])
if not pickle_headers.get():
pickle_headers.set(next(iterator))
item_count = len(pickle_headers.value)
for i in iterator:
if len(i) != item_count:
raise ValueError('Received an object with %d items, expecting %d.' % (len(i), item_count, ))
yield self.get_output(dict(zip(i)) if is_dict else dict(zip(pickle_headers.value, i)))
class PickleWriter(IOFormatEnabled, FileWriter, PickleHandler):
mode = Option(str, default='wb')
def write(self, fs, file, lineno, item):
"""
Write a pickled item to the opened file.
"""
file.write(pickle.dumps(item))
lineno += 1
return NOT_MODIFIED

0
bonobo/ext/edgy/__init__.py → bonobo/nodes/io/xml.py
View File

55
bonobo/nodes/throttle.py Normal file
View File

@ -0,0 +1,55 @@
import threading
import time
from bonobo.config import Configurable, ContextProcessor, Method, Option
class RateLimitBucket(threading.Thread):
daemon = True
@property
def stopped(self):
return self._stop_event.is_set()
def __init__(self, initial=1, period=1, amount=1):
super(RateLimitBucket, self).__init__()
self.semaphore = threading.BoundedSemaphore(initial)
self.amount = amount
self.period = period
self._stop_event = threading.Event()
def stop(self):
self._stop_event.set()
def run(self):
while not self.stopped:
time.sleep(self.period)
for _ in range(self.amount):
self.semaphore.release()
def wait(self):
return self.semaphore.acquire()
class RateLimited(Configurable):
handler = Method()
initial = Option(int, positional=True, default=1)
period = Option(int, positional=True, default=1)
amount = Option(int, positional=True, default=1)
@ContextProcessor
def bucket(self, context):
print(context)
bucket = RateLimitBucket(self.initial, self.amount, self.period)
bucket.start()
print(bucket)
yield bucket
bucket.stop()
bucket.join()
def call(self, bucket, *args, **kwargs):
print(bucket, args, kwargs)
bucket.wait()
return self.handler(*args, **kwargs)

102
bonobo/settings.py Normal file
View File

@ -0,0 +1,102 @@
import logging
import os
from bonobo.errors import ValidationError
def to_bool(s):
if s is None:
return False
if type(s) is bool:
return s
if len(s):
if s.lower() in ('f', 'false', 'n', 'no', '0'):
return False
return True
return False
class Setting:
__all__ = {}
@classmethod
def clear_all(cls):
for setting in Setting.__all__.values():
setting.clear()
def __new__(cls, name, *args, **kwargs):
Setting.__all__[name] = super().__new__(cls)
return Setting.__all__[name]
def __init__(self, name, default=None, validator=None, formatter=None):
self.name = name
if default:
self.default = default if callable(default) else lambda: default
else:
self.default = lambda: None
self.validator = validator
self.formatter = formatter
def __repr__(self):
return '<Setting {}={!r}>'.format(self.name, self.get())
def set(self, value):
value = self.formatter(value) if self.formatter else value
if self.validator and not self.validator(value):
raise ValidationError('Invalid value {!r} for setting {}.'.format(value, self.name))
self.value = value
def get(self):
try:
return self.value
except AttributeError:
value = os.environ.get(self.name, None)
if value is None:
value = self.default()
self.set(value)
return self.value
def clear(self):
try:
del self.value
except AttributeError:
pass
# Debug/verbose mode.
DEBUG = Setting('DEBUG', formatter=to_bool, default=False)
# Profile mode.
PROFILE = Setting('PROFILE', formatter=to_bool, default=False)
# Quiet mode.
QUIET = Setting('QUIET', formatter=to_bool, default=False)
# Logging level.
LOGGING_LEVEL = Setting(
'LOGGING_LEVEL',
formatter=logging._checkLevel,
validator=logging._checkLevel,
default=lambda: logging.DEBUG if DEBUG.get() else logging.INFO
)
# Input/Output format for transformations
IOFORMAT_ARG0 = 'arg0'
IOFORMAT_KWARGS = 'kwargs'
IOFORMATS = {
IOFORMAT_ARG0,
IOFORMAT_KWARGS,
}
IOFORMAT = Setting('IOFORMAT', default=IOFORMAT_KWARGS, validator=IOFORMATS.__contains__)
def check():
if DEBUG.get() and QUIET.get():
raise RuntimeError('I cannot be verbose and quiet at the same time.')
clear_all = Setting.clear_all

18
bonobo/strategies/executor.py
View File

@ -1,6 +1,5 @@
import time
import traceback
from concurrent.futures import Executor, ProcessPoolExecutor, ThreadPoolExecutor
from bonobo.constants import BEGIN, END
@ -31,12 +30,11 @@ class ExecutorStrategy(Strategy):
for plugin_context in context.plugins:
def _runner(plugin_context=plugin_context):
try:
plugin_context.start()
plugin_context.loop()
plugin_context.stop()
except Exception as exc:
print_error(exc, traceback.format_exc(), prefix='Error in plugin context', context=plugin_context)
with plugin_context:
try:
plugin_context.loop()
except Exception as exc:
print_error(exc, traceback.format_exc(), context=plugin_context)
futures.append(executor.submit(_runner))
@ -46,9 +44,7 @@ class ExecutorStrategy(Strategy):
try:
node_context.start()
except Exception as exc:
print_error(
exc, traceback.format_exc(), prefix='Could not start node context', context=node_context
)
print_error(exc, traceback.format_exc(), context=node_context, method='start')
node_context.input.on_end()
else:
node_context.loop()
@ -56,7 +52,7 @@ class ExecutorStrategy(Strategy):
try:
node_context.stop()
except Exception as exc:
print_error(exc, traceback.format_exc(), prefix='Could not stop node context', context=node_context)
print_error(exc, traceback.format_exc(), context=node_context, method='stop')
futures.append(executor.submit(_runner))

18
bonobo/structs/bags.py
View File

@ -43,7 +43,7 @@ class Bag:
def args(self):
if self._parent is None:
return self._args
return (*self._parent.args, *self._args,)
return (*self._parent.args, *self._args, )
@property
def kwargs(self):
@ -67,9 +67,7 @@ class Bag:
iter(func_or_iter)
def generator():
nonlocal func_or_iter
for x in func_or_iter:
yield x
yield from func_or_iter
return generator()
except TypeError as exc:
@ -77,6 +75,14 @@ class Bag:
raise TypeError('Could not apply bag to {}.'.format(func_or_iter))
def get(self):
"""
Get a 2 element tuple of this bag's args and kwargs.
:return: tuple
"""
return self.args, self.kwargs
def extend(self, *args, **kwargs):
return type(self)(*args, _parent=self, **kwargs)
@ -85,7 +91,7 @@ class Bag:
@classmethod
def inherit(cls, *args, **kwargs):
return cls(*args, _flags=(INHERIT_INPUT,), **kwargs)
return cls(*args, _flags=(INHERIT_INPUT, ), **kwargs)
def __eq__(self, other):
return isinstance(other, Bag) and other.args == self.args and other.kwargs == self.kwargs
@ -93,7 +99,7 @@ class Bag:
def __repr__(self):
return '<{} ({})>'.format(
type(self).__name__, ', '.
join(itertools.chain(
join(itertools.chain(
map(repr, self.args),
('{}={}'.format(k, repr(v)) for k, v in self.kwargs.items()),
))

129
bonobo/structs/graphs.py
View File

@ -1,33 +1,128 @@
from copy import copy
from bonobo.constants import BEGIN
class Graph:
"""
Represents a coherent directed acyclic graph of components.
Represents a directed graph of nodes.
"""
def __init__(self, *chain):
self.edges = {BEGIN: set()}
self.named = {}
self.nodes = []
self.graph = {BEGIN: set()}
self.add_chain(*chain)
def outputs_of(self, idx, create=False):
if create and not idx in self.graph:
self.graph[idx] = set()
return self.graph[idx]
def add_node(self, c):
i = len(self.nodes)
self.nodes.append(c)
return i
def add_chain(self, *nodes, _input=BEGIN):
for node in nodes:
_next = self.add_node(node)
self.outputs_of(_input, create=True).add(_next)
_input = _next
def __iter__(self):
yield from self.nodes
def __len__(self):
""" Node count.
"""
return len(self.nodes)
def __getitem__(self, key):
return self.nodes[key]
def outputs_of(self, idx, create=False):
""" Get a set of the outputs for a given node index.
"""
if create and not idx in self.edges:
self.edges[idx] = set()
return self.edges[idx]
def add_node(self, c):
""" Add a node without connections in this graph and returns its index.
"""
idx = len(self.nodes)
self.edges[idx] = set()
self.nodes.append(c)
return idx
def add_chain(self, *nodes, _input=BEGIN, _output=None, _name=None):
""" Add a chain in this graph.
"""
if len(nodes):
_input = self._resolve_index(_input)
_output = self._resolve_index(_output)
for i, node in enumerate(nodes):
_next = self.add_node(node)
if not i and _name:
if _name in self.named:
raise KeyError('Duplicate name {!r} in graph.'.format(_name))
self.named[_name] = _next
self.outputs_of(_input, create=True).add(_next)
_input = _next
if _output is not None:
self.outputs_of(_input, create=True).add(_output)
if hasattr(self, '_topologcally_sorted_indexes_cache'):
del self._topologcally_sorted_indexes_cache
return self
def copy(self):
g = Graph()
g.edges = copy(self.edges)
g.named = copy(self.named)
g.nodes = copy(self.nodes)
return g
@property
def topologically_sorted_indexes(self):
"""Iterate in topological order, based on networkx's topological_sort() function.
"""
try:
return self._topologcally_sorted_indexes_cache
except AttributeError:
seen = set()
order = []
explored = set()
for i in self.edges:
if i in explored:
continue
fringe = [i]
while fringe:
w = fringe[-1] # depth first search
if w in explored: # already looked down this branch
fringe.pop()
continue
seen.add(w) # mark as seen
# Check successors for cycles and for new nodes
new_nodes = []
for n in self.outputs_of(w):
if n not in explored:
if n in seen: # CYCLE !!
raise RuntimeError("Graph contains a cycle.")
new_nodes.append(n)
if new_nodes: # Add new_nodes to fringe
fringe.extend(new_nodes)
else: # No new nodes so w is fully explored
explored.add(w)
order.append(w)
fringe.pop() # done considering this node
self._topologcally_sorted_indexes_cache = tuple(filter(lambda i: type(i) is int, reversed(order)))
return self._topologcally_sorted_indexes_cache
def _resolve_index(self, mixed):
""" Find the index based on various strategies for a node, probably an input or output of chain. Supported inputs are indexes, node values or names.
"""
if mixed is None:
return None
if type(mixed) is int or mixed in self.edges:
return mixed
if isinstance(mixed, str) and mixed in self.named:
return self.named[mixed]
if mixed in self.nodes:
return self.nodes.index(mixed)
raise ValueError('Cannot find node matching {!r}.'.format(mixed))

23
bonobo/core/inputs.py → bonobo/structs/inputs.py
View File

@ -15,11 +15,12 @@
# limitations under the License.
from abc import ABCMeta, abstractmethod
from queue import Queue
from bonobo.errors import AbstractError, InactiveWritableError, InactiveReadableError
from bonobo.constants import BEGIN, END
from bonobo.basics import noop
from bonobo.errors import AbstractError, InactiveReadableError, InactiveWritableError
from bonobo.nodes import noop
BUFFER_SIZE = 8192
@ -76,6 +77,12 @@ class Input(Queue, Readable, Writable):
return Queue.put(self, data, block, timeout)
def _decrement_runlevel(self):
if self._runlevel == 1:
self.on_finalize()
self._runlevel -= 1
self.on_end()
def get(self, block=True, timeout=None):
if not self.alive:
raise InactiveReadableError('Cannot get() on an inactive {}.'.format(Readable.__name__))
@ -83,13 +90,7 @@ 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()
self._decrement_runlevel()
if not self.alive:
raise InactiveReadableError(
@ -99,6 +100,10 @@ class Input(Queue, Readable, Writable):
return data
def shutdown(self):
while self._runlevel >= 1:
self._decrement_runlevel()
def empty(self):
self.mutex.acquire()
while self._qsize() and self.queue[0] == END:

6
bonobo/util/collections.py Normal file
View File

@ -0,0 +1,6 @@
import bisect
class sortedlist(list):
def insort(self, x):
bisect.insort(self, x)

24
bonobo/util/compat.py
View File

@ -1,31 +1,7 @@
import functools
import struct
import sys
import warnings
def is_platform_little_endian():
""" am I little endian """
return sys.byteorder == 'little'
def is_platform_windows():
return sys.platform == 'win32' or sys.platform == 'cygwin'
def is_platform_linux():
return sys.platform == 'linux2'
def is_platform_mac():
return sys.platform == 'darwin'
def is_platform_32bit():
return struct.calcsize("P") * 8 < 64
def deprecated_alias(alias, func):
@functools.wraps(func)
def new_func(*args, **kwargs):

25
bonobo/util/errors.py
View File

@ -1,5 +1,7 @@
import sys
from textwrap import indent
from bonobo import settings
from bonobo.structs.bags import ErrorBag
@ -7,7 +9,14 @@ def is_error(bag):
return isinstance(bag, ErrorBag)
def print_error(exc, trace, context=None, prefix=''):
def _get_error_message(exc):
if hasattr(exc, '__str__'):
message = str(exc)
return message[0].upper() + message[1:]
return '\n'.join(exc.args),
def print_error(exc, trace, context=None, method=None):
"""
Error handler. Whatever happens in a plugin or component, if it looks like an exception, taste like an exception
or somehow make me think it is an exception, I'll handle it.
@ -18,14 +27,20 @@ def print_error(exc, trace, context=None, prefix=''):
"""
from colorama import Fore, Style
prefix = '{}{} | {}'.format(Fore.RED, Style.BRIGHT, Style.RESET_ALL)
print(
Style.BRIGHT,
Fore.RED,
'\U0001F4A3 {}{}{}'.format(
(prefix + ': ') if prefix else '', type(exc).__name__, ' in {!r}'.format(context) if context else ''
),
type(exc).__name__,
' (in {}{})'.format(type(context).__name__, '.{}()'.format(method) if method else '') if context else '',
Style.RESET_ALL,
'\n',
indent(_get_error_message(exc), prefix + Style.BRIGHT),
Style.RESET_ALL,
sep='',
file=sys.stderr,
)
print(trace)
print(prefix, file=sys.stderr)
print(indent(trace, prefix, predicate=lambda line: True), file=sys.stderr)

17
bonobo/util/helpers.py
View File

@ -1,17 +0,0 @@
from bonobo.util.compat import deprecated
@deprecated
def console_run(*chain, output=True, plugins=None, strategy=None):
from bonobo import run
from bonobo.ext.console import ConsoleOutputPlugin
return run(*chain, plugins=(plugins or []) + [ConsoleOutputPlugin()] if output else [], strategy=strategy)
@deprecated
def jupyter_run(*chain, plugins=None, strategy=None):
from bonobo import run
from bonobo.ext.jupyter import JupyterOutputPlugin
return run(*chain, plugins=(plugins or []) + [JupyterOutputPlugin()], strategy=strategy)

116
bonobo/util/inspect.py Normal file
View File

@ -0,0 +1,116 @@
from collections import namedtuple
def isconfigurabletype(mixed):
"""
Check if the given argument is an instance of :class:`bonobo.config.ConfigurableMeta`, meaning it has all the
plumbery necessary to build :class:`bonobo.config.Configurable`-like instances.
:param mixed:
:return: bool
"""
from bonobo.config.configurables import ConfigurableMeta
return isinstance(mixed, ConfigurableMeta)
def isconfigurable(mixed):
"""
Check if the given argument is an instance of :class:`bonobo.config.Configurable`.
:param mixed:
:return: bool
"""
from bonobo.config.configurables import Configurable
return isinstance(mixed, Configurable)
def isoption(mixed):
"""
Check if the given argument is an instance of :class:`bonobo.config.Option`.
:param mixed:
:return: bool
"""
from bonobo.config.options import Option
return isinstance(mixed, Option)
def ismethod(mixed):
"""
Check if the given argument is an instance of :class:`bonobo.config.Method`.
:param mixed:
:return: bool
"""
from bonobo.config.options import Method
return isinstance(mixed, Method)
def iscontextprocessor(x):
"""
Check if the given argument is an instance of :class:`bonobo.config.ContextProcessor`.
:param mixed:
:return: bool
"""
from bonobo.config.processors import ContextProcessor
return isinstance(x, ContextProcessor)
def istype(mixed):
"""
Check if the given argument is a type object.
:param mixed:
:return: bool
"""
return isinstance(mixed, type)
ConfigurableInspection = namedtuple(
'ConfigurableInspection', [
'type',
'instance',
'options',
'processors',
'partial',
]
)
ConfigurableInspection.__enter__ = lambda self: self
ConfigurableInspection.__exit__ = lambda *exc_details: None
def inspect_node(mixed, *, _partial=None):
"""
If the given argument is somehow a :class:`bonobo.config.Configurable` object (either a subclass, an instance, or
a partially configured instance), then it will return a :class:`ConfigurableInspection` namedtuple, used to inspect
the configurable metadata (options). If you want to get the option values, you don't need this, it is only usefull
to perform introspection on a configurable.
If it's not looking like a configurable, it will raise a :class:`TypeError`.
:param mixed:
:return: ConfigurableInspection
:raise: TypeError
"""
if isconfigurabletype(mixed):
inst, typ = None, mixed
elif isconfigurable(mixed):
inst, typ = mixed, type(mixed)
elif hasattr(mixed, 'func'):
return inspect_node(mixed.func, _partial=(mixed.args, mixed.keywords))
else:
raise TypeError(
'Not a Configurable, nor a Configurable instance and not even a partially configured Configurable. Check your inputs.'
)
return ConfigurableInspection(
typ,
inst,
list(typ.__options__),
list(typ.__processors__),
_partial,
)

14
bonobo/util/iterators.py
View File

@ -1,4 +1,5 @@
""" Iterator utilities. """
import functools
def force_iterator(mixed):
@ -23,6 +24,19 @@ def ensure_tuple(tuple_or_mixed):
return (tuple_or_mixed, )
def tuplize(generator):
""" Takes a generator and make it a tuple-returning function. As a side
effect, it can also decorate any iterator-returning function to force
return value to be a tuple.
"""
@functools.wraps(generator)
def tuplized(*args, **kwargs):
return tuple(generator(*args, **kwargs))
return tuplized
def iter_if_not_sequence(mixed):
if isinstance(mixed, (dict, list, str)):
raise TypeError(type(mixed).__name__)

137
bonobo/util/objects.py
View File

@ -1,3 +1,7 @@
import functools
from functools import partial
def get_name(mixed):
try:
return mixed.__name__
@ -27,178 +31,194 @@ class ValueHolder:
"""
def __init__(self, value, *, type=None):
self.value = value
self.type = type
def __init__(self, value):
self._value = value
def __repr__(self):
return repr(self.value)
@property
def value(self):
# XXX deprecated
return self._value
def __lt__(self, other):
return self.value < other
def get(self):
return self._value
def __le__(self, other):
return self.value <= other
def set(self, new_value):
self._value = new_value
def __bool__(self):
return bool(self._value)
def __eq__(self, other):
return self.value == other
return self._value == other
def __ne__(self, other):
return self.value != other
return self._value != other
def __repr__(self):
return repr(self._value)
def __lt__(self, other):
return self._value < other
def __le__(self, other):
return self._value <= other
def __gt__(self, other):
return self.value > other
return self._value > other
def __ge__(self, other):
return self.value >= other
return self._value >= other
def __add__(self, other):
return self.value + other
return self._value + other
def __radd__(self, other):
return other + self.value
return other + self._value
def __iadd__(self, other):
self.value += other
self._value += other
return self
def __sub__(self, other):
return self.value - other
return self._value - other
def __rsub__(self, other):
return other - self.value
return other - self._value
def __isub__(self, other):
self.value -= other
self._value -= other
return self
def __mul__(self, other):
return self.value * other
return self._value * other
def __rmul__(self, other):
return other * self.value
return other * self._value
def __imul__(self, other):
self.value *= other
self._value *= other
return self
def __matmul__(self, other):
return self.value @ other
return self._value @ other
def __rmatmul__(self, other):
return other @ self.value
return other @ self._value
def __imatmul__(self, other):
self.value @= other
self._value @= other
return self
def __truediv__(self, other):
return self.value / other
return self._value / other
def __rtruediv__(self, other):
return other / self.value
return other / self._value
def __itruediv__(self, other):
self.value /= other
self._value /= other
return self
def __floordiv__(self, other):
return self.value // other
return self._value // other
def __rfloordiv__(self, other):
return other // self.value
return other // self._value
def __ifloordiv__(self, other):
self.value //= other
self._value //= other
return self
def __mod__(self, other):
return self.value % other
return self._value % other
def __rmod__(self, other):
return other % self.value
return other % self._value
def __imod__(self, other):
self.value %= other
self._value %= other
return self
def __divmod__(self, other):
return divmod(self.value, other)
return divmod(self._value, other)
def __rdivmod__(self, other):
return divmod(other, self.value)
return divmod(other, self._value)
def __pow__(self, other):
return self.value**other
return self._value**other
def __rpow__(self, other):
return other**self.value
return other**self._value
def __ipow__(self, other):
self.value **= other
self._value **= other
return self
def __lshift__(self, other):
return self.value << other
return self._value << other
def __rlshift__(self, other):
return other << self.value
return other << self._value
def __ilshift__(self, other):
self.value <<= other
self._value <<= other
return self
def __rshift__(self, other):
return self.value >> other
return self._value >> other
def __rrshift__(self, other):
return other >> self.value
return other >> self._value
def __irshift__(self, other):
self.value >>= other
self._value >>= other
return self
def __and__(self, other):
return self.value & other
return self._value & other
def __rand__(self, other):
return other & self.value
return other & self._value
def __iand__(self, other):
self.value &= other
self._value &= other
return self
def __xor__(self, other):
return self.value ^ other
return self._value ^ other
def __rxor__(self, other):
return other ^ self.value
return other ^ self._value
def __ixor__(self, other):
self.value ^= other
self._value ^= other
return self
def __or__(self, other):
return self.value | other
return self._value | other
def __ror__(self, other):
return other | self.value
return other | self._value
def __ior__(self, other):
self.value |= other
self._value |= other
return self
def __neg__(self):
return -self.value
return -self._value
def __pos__(self):
return +self.value
return +self._value
def __abs__(self):
return abs(self.value)
return abs(self._value)
def __invert__(self):
return ~self.value
return ~self._value
def __len__(self):
return len(self._value)
def get_attribute_or_create(obj, attr, default):
@ -207,4 +227,3 @@ def get_attribute_or_create(obj, attr, default):
except AttributeError:
setattr(obj, attr, default)
return getattr(obj, attr)

8
bonobo/util/pkgs.py Normal file
View File

@ -0,0 +1,8 @@
import pkg_resources
from packaging.utils import canonicalize_name
bonobo_packages = {}
for p in pkg_resources.working_set:
name = canonicalize_name(p.project_name)
if name.startswith('bonobo'):
bonobo_packages[name] = p

22
bonobo/util/python.py Normal file
View File

@ -0,0 +1,22 @@
import inspect
import os
import runpy
class _RequiredModule:
def __init__(self, dct):
self.__dict__ = dct
class _RequiredModulesRegistry(dict):
def require(self, name):
if name not in self:
bits = name.split('.')
pathname = os.path.join(os.getcwd(), os.path.dirname(inspect.getfile(inspect.stack()[1][0])))
filename = os.path.join(pathname, *bits[:-1], bits[-1] + '.py')
self[name] = _RequiredModule(runpy.run_path(filename, run_name=name))
return self[name]
registry = _RequiredModulesRegistry()
require = registry.require

0
bonobo/core/statistics.py → bonobo/util/statistics.py
View File

28
bonobo/util/testing.py
View File

@ -1,5 +1,7 @@
from contextlib import contextmanager
from unittest.mock import MagicMock
from bonobo import open_fs
from bonobo.execution.node import NodeExecutionContext
@ -7,3 +9,29 @@ class CapturingNodeExecutionContext(NodeExecutionContext):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.send = MagicMock()
@contextmanager
def optional_contextmanager(cm, *, ignore=False):
if cm is None or ignore:
yield
else:
with cm:
yield
class FilesystemTester:
def __init__(self, extension='txt', mode='w'):
self.extension = extension
self.input_data = ''
self.mode = mode
def get_services_for_reader(self, tmpdir):
fs, filename = open_fs(tmpdir), 'input.' + self.extension
with fs.open(filename, self.mode) as fp:
fp.write(self.input_data)
return fs, filename, {'fs': fs}
def get_services_for_writer(self, tmpdir):
fs, filename = open_fs(tmpdir), 'output.' + self.extension
return fs, filename, {'fs': fs}