Config API

The Config API, located under the bonobo.config namespace, contains all the tools you need to create configurable transformations, either class-based or function-based.

class bonobo.config.Configurable(*args, **kwargs)[source]

Bases: object

Generic class for configurable objects. Configurable objects have a dictionary of “options” descriptors that defines the configuration schema of the type.

call(*args, **kwargs)[source]
class bonobo.config.Container[source]

Bases: dict

args_for(mixed)[source]
get(name, default=None)[source]
class bonobo.config.ContextProcessor(func)[source]

Bases: bonobo.config.options.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()
classmethod decorate(cls_or_func)[source]
class bonobo.config.Exclusive(wrapped)[source]

Bases: contextlib.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.

get_lock()[source]
class bonobo.config.Method[source]

Bases: bonobo.config.options.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()
    
clean(value)[source]
class bonobo.config.Option(type=None, *, required=False, positional=False, default=None)[source]

Bases: object

An Option is a descriptor for Configurable’s parameters.

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)

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.

(default: False)

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)

default

Default value for non-required options.

(default: None)

Example:

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')
clean(value)[source]
get_default()[source]
class bonobo.config.Service(name)[source]

Bases: bonobo.config.options.Option

A Service is a special kind of option defining a dependency to something that will be resolved at runtime, using an identifier. For example, you can create a Configurable that has a “database” Service in its attribute, meaning that you’ll define which database to use, by name, when creating the instance of this class, then provide an implementation when running the graph using a strategy.

Example:

import bonobo

class QueryExtractor(bonobo.Configurable):
    database = bonobo.Service(default='sqlalchemy.engine.default')
    
graph = bonobo.Graph(
    QueryExtractor(database='sqlalchemy.engine.secondary'),
    *more_transformations,
)

if __name__ == '__main__':
    engine = create_engine('... dsn ...')
    bonobo.run(graph, services={
        'sqlalchemy.engine.secondary': engine
    })

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).

name

Service name will be used to retrieve the implementation at runtime.

resolve(inst, services)[source]
bonobo.config.requires(*service_names)[source]