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

[core] Refactoring of commands to move towards a more pythonic way of running the jobs. Commands are now classes, and bonobo "graph" related commands now hooks into bonobo.run() calls so it will use what you actually put in your __main__ block.

Browse Source
This commit is contained in:
Romain Dorgueil
2017-10-29 19:23:50 +01:00
parent cac6920040
commit 8351897e3a
26 changed files with 483 additions and 371 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
MANIFEST.in
View File

@ -1 +1,2 @@
include *.txt
include *.py-tpl

2
Makefile
View File

@ -1,4 +1,4 @@
# Generated by Medikit 0.4a5 on 2017-10-28.
# Generated by Medikit 0.4a5 on 2017-10-29.
# All changes will be overriden.
PACKAGE ?= bonobo

23
Projectfile
View File

@ -29,24 +29,25 @@ python.setup(
'bonobo = bonobo.commands:entrypoint',
],
'bonobo.commands': [
'convert = bonobo.commands.convert:register',
'init = bonobo.commands.init:register',
'inspect = bonobo.commands.inspect:register',
'run = bonobo.commands.run:register',
'version = bonobo.commands.version:register',
'download = bonobo.commands.download:register',
'convert = bonobo.commands.convert:ConvertCommand',
'init = bonobo.commands.init:InitCommand',
'inspect = bonobo.commands.inspect:InspectCommand',
'run = bonobo.commands.run:RunCommand',
'version = bonobo.commands.version:VersionCommand',
'download = bonobo.commands.download:DownloadCommand',
],
}
)
python.add_requirements(
'colorama >=0.3,<1.0',
'fs >=2.0,<3.0',
'colorama >=0.3,<0.4',
'fs >=2.0,<2.1',
'jinja2 >=2.9,<2.10',
'packaging >=16,<17',
'psutil >=5.2,<6.0',
'psutil >=5.4,<6.0',
'python-dotenv >=0.7,<0.8',
'requests >=2.0,<3.0',
'stevedore >=1.21,<2.0',
'python-dotenv >=0.7.1,<1.0',
'stevedore >=1.27,<1.28',
dev=[
'cookiecutter >=1.5,<1.6',
'pytest-sugar >=0.8,<0.9',

7
bonobo/_api.py
View File

@ -1,5 +1,3 @@
import logging
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.nodes import LdjsonReader, LdjsonWriter
@ -21,7 +19,7 @@ def register_api_group(*args):
@register_api
def run(graph, strategy=None, plugins=None, services=None):
def run(graph, *, plugins=None, services=None, **options):
"""
Main entry point of bonobo. It takes a graph and creates all the necessary plumbery around to execute it.
@ -41,7 +39,7 @@ def run(graph, strategy=None, plugins=None, services=None):
:param dict services: The implementations of services this graph will use.
:return bonobo.execution.graph.GraphExecutionContext:
"""
strategy = create_strategy(strategy)
strategy = create_strategy(options.pop('strategy', None))
plugins = plugins or []
@ -58,6 +56,7 @@ def run(graph, strategy=None, plugins=None, services=None):
try:
from bonobo.ext.jupyter import JupyterOutputPlugin
except ImportError:
import logging
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 '

166
bonobo/commands/__init__.py
View File

@ -1,10 +1,97 @@
import argparse
import codecs
import os
import os.path
import runpy
from contextlib import contextmanager
from functools import partial
from bonobo import logging, settings
from bonobo import settings, logging
from bonobo.constants import DEFAULT_SERVICES_FILENAME, DEFAULT_SERVICES_ATTR
from bonobo.util import get_name
logger = logging.get_logger()
class BaseCommand:
@property
def logger(self):
try:
return self._logger
except AttributeError:
self._logger = logging.get_logger(get_name(self))
return self._logger
def add_arguments(self, parser):
"""
Entry point for subclassed commands to add custom arguments.
"""
pass
def handle(self, *args, **options):
"""
The actual logic of the command. Subclasses must implement this method.
"""
raise NotImplementedError('Subclasses of BaseCommand must provide a handle() method')
class BaseGraphCommand(BaseCommand):
required = True
def add_arguments(self, parser):
# target arguments (cannot provide both).
source_group = parser.add_mutually_exclusive_group(required=self.required)
source_group.add_argument('file', nargs='?', type=str)
source_group.add_argument('-m', dest='mod', type=str)
# arguments to enforce system environment.
parser.add_argument('--default-env-file', action='append')
parser.add_argument('--default-env', action='append')
parser.add_argument('--env-file', action='append')
parser.add_argument('--env', '-e', action='append')
return parser
def _run_path(self, file):
return runpy.run_path(file, run_name='__main__')
def _run_module(self, mod):
return runpy.run_module(mod, run_name='__main__')
def read(self, *, file, mod, **options):
"""
get_default_services(
filename, context.get(DEFAULT_SERVICES_ATTR)() if DEFAULT_SERVICES_ATTR in context else None
)
"""
_graph, _options = None, None
def _record(graph, **options):
nonlocal _graph, _options
_graph, _options = graph, options
with _override_runner(_record), _override_environment():
if file:
self._run_path(file)
elif mod:
self._run_module(mod)
else:
raise RuntimeError('No target provided.')
if _graph is None:
raise RuntimeError('Could not find graph.')
return _graph, _options
def handle(self, *args, **options):
pass
def entrypoint(args=None):
parser = argparse.ArgumentParser()
parser.add_argument('--debug', '-D', action='store_true')
@ -17,7 +104,15 @@ def entrypoint(args=None):
def register_extension(ext, commands=commands):
try:
parser = subparsers.add_parser(ext.name)
commands[ext.name] = ext.plugin(parser)
if isinstance(ext.plugin, type) and issubclass(ext.plugin, BaseCommand):
# current way, class based.
cmd = ext.plugin()
cmd.add_arguments(parser)
cmd.__name__ = ext.name
commands[ext.name] = cmd.handle
else:
# old school, function based.
commands[ext.name] = ext.plugin(parser)
except Exception:
logger.exception('Error while loading command {}.'.format(ext.name))
@ -33,3 +128,70 @@ def entrypoint(args=None):
logger.debug('Command: ' + args['command'] + ' Arguments: ' + repr(args))
commands[args.pop('command')](**args)
@contextmanager
def _override_runner(runner):
import bonobo
_runner_backup = bonobo.run
try:
bonobo.run = runner
yield runner
finally:
bonobo.run = _runner_backup
@contextmanager
def _override_environment(root_dir=None, **options):
yield
return
if default_env_file:
for f in default_env_file:
env_file_path = str(env_dir.joinpath(f))
load_dotenv(env_file_path)
if default_env:
for e in default_env:
set_env_var(e)
if env_file:
for f in env_file:
env_file_path = str(env_dir.joinpath(f))
load_dotenv(env_file_path, override=True)
if env:
for e in env:
set_env_var(e, override=True)
def get_default_services(filename, services=None):
dirname = os.path.dirname(filename)
services_filename = os.path.join(dirname, DEFAULT_SERVICES_FILENAME)
if os.path.exists(services_filename):
with open(services_filename) as file:
code = compile(file.read(), services_filename, 'exec')
context = {
'__name__': '__services__',
'__file__': services_filename,
}
exec(code, context)
return {
**context[DEFAULT_SERVICES_ATTR](),
**(services or {}),
}
return services or {}
def set_env_var(e, override=False):
__escape_decoder = codecs.getdecoder('unicode_escape')
ename, evalue = e.split('=', 1)
def decode_escaped(escaped):
return __escape_decoder(escaped)[0]
if len(evalue) > 0:
if evalue[0] == evalue[len(evalue) - 1] in ['"', "'"]:
evalue = decode_escaped(evalue[1:-1])
if override:
os.environ[ename] = evalue
else:
os.environ.setdefault(ename, evalue)

138
bonobo/commands/convert.py
View File

@ -1,83 +1,75 @@
import bonobo
from bonobo.commands import BaseCommand
from bonobo.registry import READER, WRITER, default_registry
from bonobo.util.resolvers import _resolve_transformations, _resolve_options
def execute(
input_filename,
output_filename,
reader=None,
reader_option=None,
writer=None,
writer_option=None,
option=None,
transformation=None,
):
reader_factory = default_registry.get_reader_factory_for(input_filename, format=reader)
reader_options = _resolve_options((option or []) + (reader_option or []))
class ConvertCommand(BaseCommand):
def add_arguments(self, parser):
parser.add_argument('input-filename', help='Input filename.')
parser.add_argument('output-filename', help='Output filename.')
parser.add_argument(
'--' + READER,
'-r',
help='Choose the reader factory if it cannot be detected from extension, or if detection is wrong.'
)
parser.add_argument(
'--' + WRITER,
'-w',
help=
'Choose the writer factory if it cannot be detected from extension, or if detection is wrong (use - for console pretty print).'
)
parser.add_argument(
'--transformation',
'-t',
dest='transformation',
action='append',
help='Add a transformation between input and output (can be used multiple times, order is preserved).',
)
parser.add_argument(
'--option',
'-O',
dest='option',
action='append',
help='Add a named option to both reader and writer factories (i.e. foo="bar").',
)
parser.add_argument(
'--' + READER + '-option',
'-' + READER[0].upper(),
dest=READER + '_option',
action='append',
help='Add a named option to the reader factory.',
)
parser.add_argument(
'--' + WRITER + '-option',
'-' + WRITER[0].upper(),
dest=WRITER + '_option',
action='append',
help='Add a named option to the writer factory.',
)
if output_filename == '-':
writer_factory = bonobo.PrettyPrinter
else:
writer_factory = default_registry.get_writer_factory_for(output_filename, format=writer)
writer_options = _resolve_options((option or []) + (writer_option or []))
def handle(self, input_filename, output_filename, reader=None, reader_option=None, writer=None, writer_option=None,
option=None, transformation=None):
reader_factory = default_registry.get_reader_factory_for(input_filename, format=reader)
reader_options = _resolve_options((option or []) + (reader_option or []))
transformations = _resolve_transformations(transformation)
if output_filename == '-':
writer_factory = bonobo.PrettyPrinter
else:
writer_factory = default_registry.get_writer_factory_for(output_filename, format=writer)
writer_options = _resolve_options((option or []) + (writer_option or []))
graph = bonobo.Graph()
graph.add_chain(
reader_factory(input_filename, **reader_options),
*transformations,
writer_factory(output_filename, **writer_options),
)
transformations = _resolve_transformations(transformation)
return bonobo.run(
graph, services={
'fs': bonobo.open_fs(),
}
)
graph = bonobo.Graph()
graph.add_chain(
reader_factory(input_filename, **reader_options),
*transformations,
writer_factory(output_filename, **writer_options),
)
def register(parser):
parser.add_argument('input-filename', help='Input filename.')
parser.add_argument('output-filename', help='Output filename.')
parser.add_argument(
'--' + READER,
'-r',
help='Choose the reader factory if it cannot be detected from extension, or if detection is wrong.'
)
parser.add_argument(
'--' + WRITER,
'-w',
help=
'Choose the writer factory if it cannot be detected from extension, or if detection is wrong (use - for console pretty print).'
)
parser.add_argument(
'--transformation',
'-t',
dest='transformation',
action='append',
help='Add a transformation between input and output (can be used multiple times, order is preserved).',
)
parser.add_argument(
'--option',
'-O',
dest='option',
action='append',
help='Add a named option to both reader and writer factories (i.e. foo="bar").',
)
parser.add_argument(
'--' + READER + '-option',
'-' + READER[0].upper(),
dest=READER + '_option',
action='append',
help='Add a named option to the reader factory.',
)
parser.add_argument(
'--' + WRITER + '-option',
'-' + WRITER[0].upper(),
dest=WRITER + '_option',
action='append',
help='Add a named option to the writer factory.',
)
return execute
return bonobo.run(
graph, services={
'fs': bonobo.open_fs(),
}
)

37
bonobo/commands/download.py
View File

@ -4,36 +4,31 @@ import re
import requests
import bonobo
from bonobo.commands import BaseCommand
EXAMPLES_BASE_URL = 'https://raw.githubusercontent.com/python-bonobo/bonobo/master/bonobo/examples/'
"""The URL to our git repository, in raw mode."""
def _write_response(response, fout):
"""Read the response and write it to the output stream in chunks."""
for chunk in response.iter_content(io.DEFAULT_BUFFER_SIZE):
fout.write(chunk)
class DownloadCommand(BaseCommand):
def handle(self, *, path, **options):
path = path.lstrip('/')
if not path.startswith('examples'):
raise ValueError('Download command currently supports examples only')
examples_path = re.sub('^examples/', '', path)
output_path = bonobo.get_examples_path(examples_path)
with _open_url(EXAMPLES_BASE_URL + examples_path) as response, open(output_path, 'wb') as fout:
for chunk in response.iter_content(io.DEFAULT_BUFFER_SIZE):
fout.write(chunk)
self.logger.info('Download saved to {}'.format(output_path))
def add_arguments(self, parser):
parser.add_argument('path', help='The relative path of the thing to download.')
def _open_url(url):
"""Open a HTTP connection to the URL and return a file-like object."""
response = requests.get(url, stream=True)
if response.status_code != 200:
raise IOError('unable to download {}, HTTP {}'.format(url, response.status_code))
raise IOError('Unable to download {}, HTTP {}'.format(url, response.status_code))
return response
def execute(path, *args, **kwargs):
path = path.lstrip('/')
if not path.startswith('examples'):
raise ValueError('download command currently supports examples only')
examples_path = re.sub('^examples/', '', path)
output_path = bonobo.get_examples_path(examples_path)
with _open_url(EXAMPLES_BASE_URL + examples_path) as response, open(output_path, 'wb') as fout:
_write_response(response, fout)
print('saved to {}'.format(output_path))
def register(parser):
parser.add_argument('path', help='The relative path of the thing to download.')
return execute

51
bonobo/commands/init.py
View File

@ -1,28 +1,33 @@
import os
def execute(name, branch):
try:
from cookiecutter.main import cookiecutter
except ImportError as exc:
raise ImportError(
'You must install "cookiecutter" to use this command.\n\n $ pip install cookiecutter\n'
) from exc
from jinja2 import Environment, FileSystemLoader
overwrite_if_exists = False
project_path = os.path.join(os.getcwd(), name)
if os.path.isdir(project_path) and not os.listdir(project_path):
overwrite_if_exists = True
return cookiecutter(
'https://github.com/python-bonobo/cookiecutter-bonobo.git',
extra_context={'name': name},
no_input=True,
checkout=branch,
overwrite_if_exists=overwrite_if_exists
)
from bonobo.commands import BaseCommand
def register(parser):
parser.add_argument('name')
parser.add_argument('--branch', '-b', default='master')
return execute
class InitCommand(BaseCommand):
TEMPLATES = {'job'}
TEMPLATES_PATH = os.path.join(os.path.dirname(__file__), 'templates')
def add_arguments(self, parser):
parser.add_argument('template', choices=self.TEMPLATES)
parser.add_argument('filename')
parser.add_argument('--force', '-f', default=False, action='store_true')
def handle(self, *, template, filename, force=False):
template_name = template
name, ext = os.path.splitext(filename)
if ext != '.py':
raise ValueError('Filenames should end with ".py".')
loader = FileSystemLoader(self.TEMPLATES_PATH)
env = Environment(loader=loader)
template = env.get_template(template_name + '.py-tpl')
if os.path.exists(filename) and not force:
raise FileExistsError('Target filename already exists, use --force to override.')
with open(filename, 'w+') as f:
f.write(template.render(name=name))
self.logger.info('Generated {} using template {!r}.'.format(filename, template_name))

47
bonobo/commands/inspect.py
View File

@ -1,40 +1,21 @@
import json
from bonobo.commands import BaseGraphCommand
from bonobo.commands.run import read, register_generic_run_arguments
from bonobo.constants import BEGIN
from bonobo.util.objects import get_name
OUTPUT_GRAPHVIZ = 'graphviz'
OUTPUT_GRAPH = 'graphviz'
def _ident(graph, i):
escaped_index = str(i)
escaped_name = json.dumps(get_name(graph[i]))
return '{{{} [label={}]}}'.format(escaped_index, escaped_name)
class InspectCommand(BaseGraphCommand):
def add_arguments(self, parser):
super(InspectCommand, self).add_arguments(parser)
parser.add_argument('--graph', '-g', dest='output', action='store_const', const=OUTPUT_GRAPH)
def handle(self, output=None, **options):
if output is None:
raise ValueError('Output type must be provided (try --graph/-g).')
def execute(*, output, **kwargs):
graph, plugins, services = read(**kwargs)
graph, params = self.read(**options)
if output == OUTPUT_GRAPHVIZ:
print('digraph {')
print(' rankdir = LR;')
print(' "BEGIN" [shape="point"];')
if output == OUTPUT_GRAPH:
print(graph._repr_dot_())
else:
raise NotImplementedError('Output type not implemented.')
for i in graph.outputs_of(BEGIN):
print(' "BEGIN" -> ' + _ident(graph, i) + ';')
for ix in graph.topologically_sorted_indexes:
for iy in graph.outputs_of(ix):
print(' {} -> {};'.format(_ident(graph, ix), _ident(graph, iy)))
print('}')
else:
raise NotImplementedError('Output type not implemented.')
def register(parser):
register_generic_run_arguments(parser)
parser.add_argument('--graph', '-g', dest='output', action='store_const', const=OUTPUT_GRAPHVIZ)
parser.set_defaults(output=OUTPUT_GRAPHVIZ)
return execute

215
bonobo/commands/run.py
View File

@ -1,38 +1,60 @@
import codecs
import os
import sys
from importlib.util import spec_from_file_location, module_from_spec
from pathlib import Path
from dotenv import load_dotenv
import bonobo
from bonobo.constants import DEFAULT_SERVICES_ATTR, DEFAULT_SERVICES_FILENAME
DEFAULT_GRAPH_FILENAMES = (
'__main__.py',
'main.py',
)
DEFAULT_GRAPH_ATTR = 'get_graph'
from bonobo.commands import BaseGraphCommand
def get_default_services(filename, services=None):
dirname = os.path.dirname(filename)
services_filename = os.path.join(dirname, DEFAULT_SERVICES_FILENAME)
if os.path.exists(services_filename):
with open(services_filename) as file:
code = compile(file.read(), services_filename, 'exec')
context = {
'__name__': '__bonobo__',
'__file__': services_filename,
}
exec(code, context)
class RunCommand(BaseGraphCommand):
install = False
return {
**context[DEFAULT_SERVICES_ATTR](),
**(services or {}),
}
return services or {}
def add_arguments(self, parser):
super(RunCommand, self).add_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')
def _run_path(self, file):
if self.install:
if os.path.isdir(file):
requirements = os.path.join(file, 'requirements.txt')
else:
requirements = os.path.join(os.path.dirname(file), 'requirements.txt')
_install_requirements(requirements)
return super()._run_path(file)
def _run_module(self, mod):
if self.install:
raise RuntimeError('--install behaviour when running a module is not defined.')
return super()._run_module(mod)
def handle(self, *args, quiet=False, verbose=False, install=False, **options):
from bonobo import settings
settings.QUIET.set_if_true(quiet)
settings.DEBUG.set_if_true(verbose)
self.install = install
graph, params = self.read(**options)
params['plugins'] = set(params.pop('plugins', ())).union(set(options.pop('plugins', ())))
return bonobo.run(graph, **params)
def register_generic_run_arguments(parser, required=True):
"""
Only there for backward compatibility with third party extensions.
TODO: This should be deprecated (using the @deprecated decorator) in 0.7, and removed in 0.8 or 0.9.
"""
dummy_command = BaseGraphCommand()
dummy_command.required = required
dummy_command.add_arguments(parser)
return parser
def _install_requirements(requirements):
@ -47,138 +69,3 @@ def _install_requirements(requirements):
pip.utils.pkg_resources = importlib.reload(pip.utils.pkg_resources)
import site
importlib.reload(site)
def read(
filename,
module,
install=False,
quiet=False,
verbose=False,
default_env_file=None,
default_env=None,
env_file=None,
env=None
):
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)
spec = spec_from_file_location('__bonobo__', filename)
main = sys.modules['__bonobo__'] = module_from_spec(spec)
main.__path__ = [os.path.dirname(filename)]
main.__package__ = '__bonobo__'
spec.loader.exec_module(main)
context = main.__dict__
elif module:
context = runpy.run_module(module, run_name='__bonobo__')
filename = context['__file__']
else:
raise RuntimeError('UNEXPECTED: argparse should not allow this.')
env_dir = Path(filename).parent or Path(module).parent
if default_env_file:
for f in default_env_file:
env_file_path = str(env_dir.joinpath(f))
load_dotenv(env_file_path)
if default_env:
for e in default_env:
set_env_var(e)
if env_file:
for f in env_file:
env_file_path = str(env_dir.joinpath(f))
load_dotenv(env_file_path, override=True)
if env:
for e in env:
set_env_var(e, override=True)
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, '
'but it is something that will be implemented in the future.\n\nExpected: 1, got: {}.'
).format(len(graphs))
graph = list(graphs.values())[0]
plugins = []
services = get_default_services(
filename, context.get(DEFAULT_SERVICES_ATTR)() if DEFAULT_SERVICES_ATTR in context else None
)
return graph, plugins, services
def set_env_var(e, override=False):
__escape_decoder = codecs.getdecoder('unicode_escape')
ename, evalue = e.split('=', 1)
def decode_escaped(escaped):
return __escape_decoder(escaped)[0]
if len(evalue) > 0:
if evalue[0] == evalue[len(evalue) - 1] in ['"', "'"]:
evalue = decode_escaped(evalue[1:-1])
if override:
os.environ[ename] = evalue
else:
os.environ.setdefault(ename, evalue)
def execute(
filename,
module,
install=False,
quiet=False,
verbose=False,
default_env_file=None,
default_env=None,
env_file=None,
env=None
):
graph, plugins, services = read(
filename, module, install, quiet, verbose, default_env_file, default_env, env_file, env
)
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)
parser.add_argument('--default-env-file', action='append')
parser.add_argument('--default-env', action='append')
parser.add_argument('--env-file', action='append')
parser.add_argument('--env', '-e', action='append')
return parser
def register(parser):
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

50
bonobo/commands/templates/default.py-tpl Normal file
View File

@ -0,0 +1,50 @@
import bonobo
def extract():
"""Placeholder, change, rename, remove... """
yield 'hello'
yield 'world'
def transform(*args):
"""Placeholder, change, rename, remove... """
yield tuple(
map(str.title, args)
)
def load(*args):
"""Placeholder, change, rename, remove... """
print(*args)
def get_graph():
"""
This function builds the graph that needs to be executed.
:return: bonobo.Graph
"""
graph = bonobo.Graph()
graph.add_chain(extract, transform, load)
return graph
def get_services():
"""
This function builds the services dictionary, which is a simple dict of names-to-implementation used by bonobo
for runtime injection.
It will be used on top of the defaults provided by bonobo (fs, http, ...). You can override those defaults, or just
let the framework define them. You can also define your own services and naming is up to you.
:return: dict
"""
return {}
# The __main__ block actually execute the graph.
if __name__ == '__main__':
# Although you're not required to use it, bonobo's graph related commands will hook to this call (inspect, run, ...).
bonobo.run(get_graph(), services=get_services())

52
bonobo/commands/version.py
View File

@ -1,4 +1,30 @@
def format_version(mod, *, name=None, quiet=False):
from bonobo.commands import BaseCommand
class VersionCommand(BaseCommand):
def handle(self, *, 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 add_arguments(self, parser):
parser.add_argument('--all', '-a', action='store_true')
parser.add_argument('--quiet', '-q', action='count')
def _format_version(mod, *, name=None, quiet=False):
from bonobo.util.pkgs import bonobo_packages
args = {
'name': name or mod.__name__,
@ -14,27 +40,3 @@ def format_version(mod, *, name=None, quiet=False):
return '{version}'.format(**args)
raise RuntimeError('Hard to be so quiet...')
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

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

@ -14,7 +14,7 @@ Extracts a list of parisian bars where you can buy a coffee for a reasonable pri
"""
import bonobo
from bonobo.commands.run import get_default_services
from bonobo.commands import get_default_services
from bonobo.ext.opendatasoft import OpenDataSoftAPI
filename = 'coffeeshops.txt'

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

@ -19,7 +19,7 @@ import json
from colorama import Fore, Style
import bonobo
from bonobo.commands.run import get_default_services
from bonobo.commands import get_default_services
from bonobo.ext.opendatasoft import OpenDataSoftAPI
try:

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

@ -1,5 +1,5 @@
import bonobo
from bonobo.commands.run import get_default_services
from bonobo.commands import get_default_services
graph = bonobo.Graph(
bonobo.CsvReader('datasets/coffeeshops.txt', headers=('item', )),

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

@ -1,6 +1,6 @@
import bonobo
from bonobo import Bag
from bonobo.commands.run import get_default_services
from bonobo.commands import get_default_services
def get_fields(**row):

2
bonobo/examples/files/pickle_handlers.py
View File

@ -28,7 +28,7 @@ messages categorized as spam, and (3) prints the output.
'''
import bonobo
from bonobo.commands.run import get_default_services
from bonobo.commands import get_default_services
from fs.tarfs import TarFS

2
bonobo/examples/files/text_handlers.py
View File

@ -1,5 +1,5 @@
import bonobo
from bonobo.commands.run import get_default_services
from bonobo.commands import get_default_services
def skip_comments(line):

2
bonobo/ext/django.py
View File

@ -5,7 +5,7 @@ from django.core.management.base import BaseCommand, OutputWrapper
import bonobo
import bonobo.util
from bonobo.commands.run import get_default_services
from bonobo.commands import get_default_services
from bonobo.ext.console import ConsoleOutputPlugin
from bonobo.util.term import CLEAR_EOL

6
bonobo/settings.py
View File

@ -51,6 +51,12 @@ class Setting:
raise ValidationError('Invalid value {!r} for setting {}.'.format(value, self.name))
self.value = value
def set_if_true(self, value):
"""Sets the value to true if it is actually true. May sound strange but the main usage is enforcing some
settings from command line."""
if value:
self.set(True)
def get(self):
try:
return self.value

26
bonobo/structs/graphs.py
View File

@ -1,6 +1,8 @@
import json
from copy import copy
from bonobo.constants import BEGIN
from bonobo.util import get_name
class Graph:
@ -110,6 +112,24 @@ class Graph:
self._topologcally_sorted_indexes_cache = tuple(filter(lambda i: type(i) is int, reversed(order)))
return self._topologcally_sorted_indexes_cache
def _repr_dot_(self):
src = [
'digraph {',
' rankdir = LR;',
' "BEGIN" [shape="point"];',
]
for i in self.outputs_of(BEGIN):
src.append(' "BEGIN" -> ' + _get_graphviz_node_id(self, i) + ';')
for ix in self.topologically_sorted_indexes:
for iy in self.outputs_of(ix):
src.append(' {} -> {};'.format(_get_graphviz_node_id(self, ix), _get_graphviz_node_id(self, iy)))
src.append('}')
return '\n'.join(src)
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.
"""
@ -126,3 +146,9 @@ class Graph:
return self.nodes.index(mixed)
raise ValueError('Cannot find node matching {!r}.'.format(mixed))
def _get_graphviz_node_id(graph, i):
escaped_index = str(i)
escaped_name = json.dumps(get_name(graph[i]))
return '{{{} [label={}]}}'.format(escaped_index, escaped_name)

2
requirements-dev.txt
View File

@ -27,7 +27,7 @@ pytz==2017.2
requests==2.18.4
six==1.11.0
snowballstemmer==1.2.1
sphinx==1.6.4
sphinx==1.6.5
sphinxcontrib-websupport==1.0.1
termcolor==1.1.0
urllib3==1.22

2
requirements-docker.txt
View File

@ -3,6 +3,7 @@ appdirs==1.4.3
bonobo-docker==0.5.0
certifi==2017.7.27.1
chardet==3.0.4
click==6.7
colorama==0.3.9
docker-pycreds==0.2.1
docker==2.3.0
@ -12,6 +13,7 @@ packaging==16.8
pbr==3.1.1
psutil==5.4.0
pyparsing==2.2.0
python-dotenv==0.7.1
pytz==2017.2
requests==2.18.4
six==1.11.0

2
requirements-jupyter.txt
View File

@ -16,7 +16,7 @@ jupyter-console==5.2.0
jupyter-core==4.3.0
jupyter==1.0.0
markupsafe==1.0
mistune==0.7.4
mistune==0.8
nbconvert==5.3.1
nbformat==4.4.0
notebook==5.2.0

2
requirements.txt
View File

@ -6,6 +6,8 @@ click==6.7
colorama==0.3.9
fs==2.0.12
idna==2.6
jinja2==2.9.6
markupsafe==1.0
packaging==16.8
pbr==3.1.1
psutil==5.4.0

11
setup.py
View File

@ -53,8 +53,9 @@ setup(
packages=find_packages(exclude=['ez_setup', 'example', 'test']),
include_package_data=True,
install_requires=[
'colorama (>= 0.3, < 1.0)', 'fs (>= 2.0, < 3.0)', 'packaging (>= 16, < 17)', 'psutil (>= 5.2, < 6.0)',
'python-dotenv (>= 0.7.1, < 1.0)', 'requests (>= 2.0, < 3.0)', 'stevedore (>= 1.21, < 2.0)'
'colorama (>= 0.3, < 0.4)', 'fs (>= 2.0, < 2.1)', 'jinja2 (>= 2.9, < 2.10)', 'packaging (>= 16, < 17)',
'psutil (>= 5.4, < 6.0)', 'python-dotenv (>= 0.7, < 0.8)', 'requests (>= 2.0, < 3.0)',
'stevedore (>= 1.27, < 1.28)'
],
extras_require={
'dev': [
@ -67,9 +68,9 @@ setup(
},
entry_points={
'bonobo.commands': [
'convert = bonobo.commands.convert:register', 'init = bonobo.commands.init:register',
'inspect = bonobo.commands.inspect:register', 'run = bonobo.commands.run:register',
'version = bonobo.commands.version:register', 'download = bonobo.commands.download:register'
'convert = bonobo.commands.convert:ConvertCommand', 'init = bonobo.commands.init:InitCommand',
'inspect = bonobo.commands.inspect:InspectCommand', 'run = bonobo.commands.run:RunCommand',
'version = bonobo.commands.version:VersionCommand', 'download = bonobo.commands.download:DownloadCommand'
],
'console_scripts': ['bonobo = bonobo.commands:entrypoint']
},