b3j0f.annotation package

Module contents

Annotation package.

class b3j0f.annotation.Annotation(on_bind_target=None, propagate=True, override=False, ttl=None, in_memory=False)[source]

Bases: object

Base class for all annotations defined in this library.

It contains functions to override in order to catch initialisation of this Annotation and annotated elements binding (also called commonly target in the context of Annotation).

All annotations which inherit from this are registered to target objects and are accessibles through the static method Annotation.get_annotations.

Instance methods to override are: - __init__: set parameters during its instantiation. - _bind_target: called to bind target to this. - on_bind_target: fired when the annotated element is bound to this.

And properties are: - propagate: (default True) determines if an annotation is propagated to all sub target elements. - override: (default False) exclude previous annotation of the same type as self class. - _ttl: (default None) self time to leave. - _in_memory: (default False) save instance in a global dictionary.

It is also possible to set on_bind_target, propagate and override in
the constructor.
IN_MEMORY = '_in_memory'
OVERRIDE = 'override'
PROPAGATE = 'propagate'
TARGETS = 'targets'
TTL = '_ttl'
bind_target(target, ctx=None)[source]

Bind self annotation to target.

Parameters:
  • target – target to annotate.
  • ctx – target ctx.
Returns:

bound target.

classmethod free_memory(exclude=None)[source]

Free global annotation memory.

classmethod get_annotated_fields(instance, select=<function <lambda>>)[source]

Get dict of {annotated fields: annotations} by cls of input instance.

Returns:a set of (annotated fields, annotations).
Return type:dict
classmethod get_annotations(target, exclude=None, ctx=None, select=<function <lambda>>, mindepth=0, maxdepth=0, followannotated=True, public=True, _history=None)[source]

Returns all input target annotations of cls type sorted by definition order.

Parameters:
  • cls (type) – type of annotation to get from target.
  • target – target from where get annotations.
  • exclude (tuple/type) – annotation types to remove from selection.
  • ctx – target ctx.
  • select – bool function which select annotations after applying previous type filters. Takes a target, a ctx and an annotation in parameters. True by default.
  • mindepth (int) – minimal depth for searching annotations (default 0)
  • maxdepth (int) – maximal depth for searching annotations (default 0)
  • followannotated (bool) – if True (default) follow deeply only annotated members.
  • public (bool) – if True (default) follow public members.
  • _history (list) – private parameter which save parsed elements.
Return type:

Annotation

classmethod get_local_annotations(target, exclude=None, ctx=None, select=<function <lambda>>)[source]
Get a list of local target annotations in the order of their
definition.
Parameters:
  • cls (type) – type of annotation to get from target.
  • target – target from where get annotations.
  • exclude (tuple/type) – annotation types to exclude from selection.
  • ctx – target ctx.
  • select – selection function which takes in parameters a target, a ctx and an annotation and returns True if the annotation has to be selected. True by default.
Returns:

target local annotations.

Return type:

list

classmethod get_memory_annotations(exclude=None)[source]

Get annotations in memory which inherits from cls.

Parameters:exclude (tuple/type) – annotation type(s) to exclude from search.
Returns:found annotations which inherits from cls.
Return type:set
in_memory
Returns:True if self is in a global memory of annotations.
on_bind_target(target, ctx=None)[source]

Fired after target is bound to self.

Parameters:
  • target – newly bound target.
  • ctx – target ctx.
override
propagate
classmethod remove(target, exclude=None, ctx=None, select=<function <lambda>>)[source]

Remove from target annotations which inherit from cls.

Parameters:
  • target – target from where remove annotations which inherits from cls.
  • exclude (tuple/type) – annotation types to exclude from selection.
  • ctx – target ctx.
  • select – annotation selection function which takes in parameters a target, a ctx and an annotation and return True if the annotation has to be removed.
remove_from(target, ctx=None)[source]

Remove self annotation from target annotations.

Parameters:
  • target – target from where remove self annotation.
  • ctx – target ctx.
targets
ttl

Get actual ttl in seconds.

Returns:actual ttl.
Return type:float
class b3j0f.annotation.Synchronized(lock=None, *args, **kwargs)[source]

Bases: b3j0f.annotation.interception.PrivateInterceptor

Transform a target into a thread safe target.

enable
interception
override
propagate
targets
class b3j0f.annotation.SynchronizedClass(lock=None, *args, **kwargs)[source]

Bases: b3j0f.annotation.async.Synchronized

Transform a class into a thread safe class.

on_bind_target(target, ctx=None)[source]
class b3j0f.annotation.Asynchronous(*args, **kwargs)[source]

Bases: b3j0f.annotation.core.Annotation

Transform a target into an asynchronous callable target.

exception NotYetDoneException[source]

Bases: exceptions.Exception

Handle when a result is not yet available.

class Asynchronous.Result(queue, thread)[source]

Bases: object

In charge of receive asynchronous function result.

get_result(wait=-1)[source]

Get result value.

Wait for it if necessary.

Parameters:wait (int) – maximum wait time.
Returns:result value.
is_done()[source]

True if result is available.

queue
result
thread
Asynchronous.on_bind_target(target, ctx=None)[source]
Asynchronous.start(*args, **kwargs)[source]

Start execution of the function.

class b3j0f.annotation.TimeOut(seconds, error_message='Call of {0} with parameters {1} and {2} is timed out in frame {3}', *args, **kwargs)[source]

Bases: b3j0f.annotation.interception.PrivateInterceptor

Raise an Exception if the target call has not finished in time.

ERROR_MESSAGE = 'error_message'
SECONDS = 'seconds'
exception TimeOutError(timeout_interceptor, frame)[source]

Bases: exceptions.Exception

Exception thrown if time elapsed before the end of the target call.

DEFAULT_MESSAGE = 'Call of {0} with parameters {1} and {2} is timed out in frame {3}'
TimeOut.enable
TimeOut.error_message
TimeOut.interception
TimeOut.override
TimeOut.propagate
TimeOut.seconds
TimeOut.targets
class b3j0f.annotation.Wait(before=1, after=1, *args, **kwargs)[source]

Bases: b3j0f.annotation.interception.PrivateInterceptor

Define a time to wait before and after a target call.

AFTER = 'after'
BEFORE = 'before'
DEFAULT_AFTER = 1
DEFAULT_BEFORE = 1
after
before
enable
interception
override
propagate
targets
class b3j0f.annotation.Observable(*args, **kwargs)[source]

Bases: b3j0f.annotation.interception.PrivateInterceptor

Imlementation of the observer design pattern.

It transforms a target into an observable object in adding method register_observer, unregister_observer and notify_observers. Observers listen to pre/post target interception.

notify_observers(joinpoint, post=False)[source]

Notify observers with parameter calls and information about pre/post call.

on_bind_target(target, ctx=None)[source]
register_observer(observer)[source]

Register an observer.

unregister_observer(observer)[source]

Unregister an observer.

class b3j0f.annotation.Types(rtype=None, ptypes=None, *args, **kwargs)[source]

Bases: b3j0f.annotation.interception.PrivateInterceptor

Check routine parameters and return type.

class NamedParameterType(name, parameter_type)[source]

Bases: object

Handle Named Parameter Type.

class Types.NamedParameterTypes(target, named_parameter_types)[source]

Bases: object

Handle Named Parameter Types.

class Types.NotEmpty(_type)[source]

Bases: b3j0f.annotation.call.SpecialCondition

Handle NotEmpty SpecialCondition.

class Types.NotNone(_type)[source]

Bases: b3j0f.annotation.call.SpecialCondition

Handle NotNone SpecialCondition.

Types.PTYPES = 'ptypes'
Types.RTYPE = 'rtype'
class Types.SpecialCondition(_type)[source]

Bases: object

Handle SpecialCondition.

get_type()[source]

Get special condition parameter type.

exception Types.TypesError[source]

Bases: exceptions.Exception

Handle Types error.

Types.bind_target(target, ctx=None)

Bind self annotation to target.

Parameters:
  • target – target to annotate.
  • ctx – target ctx.
Returns:

bound target.

static Types.check_value(value, expected_type)[source]

Check Types parameters.

Types.enable
Types.interception
Types.override
Types.propagate
Types.ptypes
Types.rtype
Types.targets
b3j0f.annotation.types(*args, **kwargs)[source]

Quick alias for the Types Annotation with only args and kwargs parameters.

Parameters:
  • args (tuple) – may contain rtype.
  • kwargs (dict) – may contain ptypes.
class b3j0f.annotation.Curried(varargs=None, keywords=None, *args, **kwargs)[source]

Bases: b3j0f.annotation.interception.PrivateInterceptor

Annotation that returns a function that keeps returning functions until all arguments are supplied; then the original function is evaluated.

Inspirated from Jeff Laughlin Consulting LLC projects.

ARGS = 'args'
class CurriedResult(curried, exception)[source]

Bases: object

Curried result in case of missing arguments.

curried
exception
Curried.DEFAULT_ARGS = 'default_args'
Curried.DEFAULT_KWARGS = 'default_kwargs'
Curried.KWARGS = 'kwargs'
Curried.args
Curried.default_args
Curried.default_kwargs
Curried.enable
Curried.interception
Curried.kwargs
Curried.override
Curried.propagate
Curried.targets
b3j0f.annotation.curried(*args, **kwargs)[source]

Curried annotation with varargs and kwargs.

class b3j0f.annotation.Retries(max_tries, delay=1, backoff=2, exceptions=(<type 'exceptions.Exception'>, ), hook=None, condition=3, *args, **kwargs)[source]

Bases: b3j0f.annotation.interception.PrivateInterceptor

Function decorator implementing retrying logic.

condition: retry condition, among execution success or failure or both. delay: Sleep this many seconds * backoff * try number after failure backoff: Multiply delay by this factor after each failure exceptions: A tuple of exception classes; default (Exception,) hook: A function with the signature myhook(data, condition, tries_remaining , mydelay) where data is result function or raised Exception, condition is ON_ERROR or ON_SUCCESS depending on error or success execution function, tries_remaining is tries remaining, and finally, mydelay is waiting seconds between calls; default None.

The decorator will call the function up to max_tries times if it raises an exception or if it simply execute the function, depending on state condition.

By default it catches instances of the Exception class and subclasses. This will recover after all but the most fatal errors. You may specify a custom tuple of exception classes with the ‘exceptions’ argument; the function will only be retried if it raises one of the specified exceptions.

Additionally you may specify a hook function which will be called prior to retrying with the number of remaining tries and the exception instance; see given example. This is primarily intended to give the opportunity to log the failure. Hook is not called after failure if no retries remain.

ALL = 3
BACKOFF = 'backoff'
CONDITION = 'condition'
DEFAULT_BACKOFF = 2
DEFAULT_DELAY = 1
DEFAULT_EXCEPTIONS = (<type 'exceptions.Exception'>,)
DELAY = 'delay'
EXCEPTIONS = 'exceptions'
HOOK = 'hook'
MAX_TRIES = 'max_tries'
ON_ERROR = 1
ON_SUCCESS = 2
backoff
condition
delay
enable
exceptions
hook
interception
max_tries
override
propagate
targets
class b3j0f.annotation.Condition(pre_cond=None, post_cond=None, *args, **kwargs)[source]

Bases: b3j0f.annotation.interception.PrivateInterceptor

Apply a pre/post condition on an annotated element call.

exception ConditionError[source]

Bases: exceptions.Exception

Handle condition errors.

Condition.POST_COND = 'post_cond'
Condition.PRE_COND = 'pre_cond'
exception Condition.PostConditionError[source]

Bases: b3j0f.annotation.check.ConditionError

Handle post condition errors.

exception Condition.PreConditionError[source]

Bases: b3j0f.annotation.check.ConditionError

Handle pre condition errors.

Condition.RESULT = 'result'
Condition.enable
Condition.interception
Condition.override
Condition.post_cond
Condition.pre_cond
Condition.propagate
Condition.targets
class b3j0f.annotation.MaxCount(count=1, *args, **kwargs)[source]

Bases: b3j0f.annotation.check.AnnotationChecker

Set a maximum count of target instantiation by annotated element.

DEFAULT_COUNT = 1
exception Error[source]

Bases: exceptions.Exception

Handle MaxCount errors.

MaxCount.bind_target(target, ctx=None)

Bind self annotation to target.

Parameters:
  • target – target to annotate.
  • ctx – target ctx.
Returns:

bound target.

MaxCount.count
MaxCount.enable
MaxCount.interception
MaxCount.override
MaxCount.propagate
MaxCount.targets
class b3j0f.annotation.Target(types=None, rule='or', instances=False, *args, **kwargs)[source]

Bases: b3j0f.annotation.check.AnnotationChecker

Check type of all decorated elements by this decorated Annotation in using a list of types.

This list of types is checked related to logical rules such as OR and AND.

  • OR: true if at least one type is checked.
  • AND: true if all types are checked.
AND = 'and'
CALLABLE()

callable(object) -> bool

Return whether the object is callable (i.e., some kind of function). Note that classes are callable, as are instances with a __call__() method.

DEFAULT_INSTANCES = False
DEFAULT_RULE = 'or'
exception Error[source]

Bases: exceptions.Exception

Handle Target errors.

Target.FUNC

alias of function

Target.INSTANCES = 'instances'
Target.OR = 'or'
Target.ROUTINE = 'routine'
Target.RULE = 'rule'
Target.TYPES = 'types'
Target.bind_target(target, ctx=None)

Bind self annotation to target.

Parameters:
  • target – target to annotate.
  • ctx – target ctx.
Returns:

bound target.

Target.enable
Target.instances
Target.interception
Target.override
Target.propagate
Target.rule
Target.targets
Target.types
class b3j0f.annotation.Interceptor(interception=None, pointcut=None, enable=True, *args, **kwargs)[source]

Bases: b3j0f.annotation.core.Annotation

Annotation able to intercept annotated elements.

This interception can be disabled at any time and specialized with a
pointcut.
ENABLE = 'enable'
INTERCEPTION = 'interception'
INTERCEPTOR = '__interceptor__'
exception InterceptorError[source]

Bases: exceptions.Exception

Handle Interceptor errors.

Interceptor.POINTCUT = 'pointcut'
Interceptor.enable
Interceptor.interception
Interceptor.intercepts(joinpoint)[source]

Self target interception if self is enabled

Parameters:joinpoint – advices executor
Interceptor.override
Interceptor.pointcut

Get pointcut.

Interceptor.propagate
Interceptor.remove_from(target, ctx=None, *args, **kwargs)[source]
classmethod Interceptor.set_enable(target, enable=True)[source]

(Dis|En)able annotated interceptors.

Interceptor.targets
class b3j0f.annotation.PrivateInterceptor(*args, **kwargs)[source]

Bases: b3j0f.annotation.interception.Interceptor

Interceptor with a private interception resource.

enable
interception
override
propagate
targets
class b3j0f.annotation.CallInterceptor(*args, **kwargs)[source]

Bases: b3j0f.annotation.interception.Interceptor

Interceptor dedicated to intercept call of annotated element.

Instead of Interceptor, the pointcut equals ‘__call__’, therefore, the target may be a class instead of an instance beceause python does not handle system methods (such as __call__, __repr__, etc.) if they are overriden on instance.

It could be used to intercepts annotation binding annotated by self.

enable
interception
override
propagate
targets
class b3j0f.annotation.PrivateCallInterceptor(*args, **kwargs)[source]

Bases: b3j0f.annotation.interception.CallInterceptor

Interceptor dedicated to apply a private interception on target calls.

Instead of Interceptor, the pointcut equals ‘__call__’, therefore, the target may be a class instead of an instance beceause python does not handle system methods (such as __call__, __repr__, etc.) if they are overriden on instance.

enable
interception
override
propagate
targets
class b3j0f.annotation.Transform(name=None, bases=None, _dict=None, update=True, *args, **kwargs)[source]

Bases: b3j0f.annotation.core.Annotation

Transform a class into an annotation or something else if parameters are different.

BASES = 'bases'
DICT = 'dict'
NAME = 'name'
UPDATE = 'update'
bases
dict
name
override
propagate
targets
update
class b3j0f.annotation.Mixin(classes=(), *attributes, **named_attributes)[source]

Bases: b3j0f.annotation.core.Annotation

Annotation which enrichs a target with Mixin.

For every defined mixin, a private couple of (name, array of mixed items) is created into the target in order to go back in a no mixin state.

exception MixInError[source]

Bases: exceptions.Exception

Raised for any Mixin error.

static Mixin.get_mixedins_by_name(target)[source]

Get a set of couple (name, field) of target mixedin.

static Mixin.mixin(target, resource, name=None)[source]

Do the correct mixin depending on the type of input resource.

  • Method or Function: mixin_function_or_method.
  • class: mixin_class.
  • other: set_mixin.

And returns the result of the choosen method (one or a list of mixins).

static Mixin.mixin_class(target, cls)[source]

Mix cls content in target.

static Mixin.mixin_function_or_method(target, routine, name=None, isbound=False)[source]

Mixin a routine into the target.

Parameters:
  • routine – routine to mix in target.
  • name (str) – mixin name. Routine name by default.
  • isbound (bool) – If True (False by default), the mixin result is a bound method to target.
Mixin.on_bind_target(target, ctx=None)[source]
static Mixin.remove_mixin(target, name, mixedin=None, replace=True)[source]

Remove a mixin with name (and reference) from targetand returns the replaced one or None.

Parameters:
  • mixedin – a mixedin value or the last defined mixedin if is None (by default).
  • replace (bool) – If True (default), the removed mixedin replaces the current mixin.
static Mixin.remove_mixins(target)[source]

Tries to get back target in a no mixin consistent state.

static Mixin.set_mixin(target, resource, name=None, override=True)[source]

Set a resource and returns the mixed one in target content or Mixin.__NEW_CONTENT_KEY__ if resource name didn’t exist in target.

Parameters:
  • name (str) – target content item to mix with the resource.
  • override (bool) – If True (default) permits to replace an old resource by the new one.
class b3j0f.annotation.Deprecated(*args, **kwargs)[source]

Bases: b3j0f.annotation.interception.PrivateInterceptor

Decorator which can be used to mark functions as deprecated. It will result in a warning being emitted when the function is used.

class b3j0f.annotation.Singleton(*args, **kwargs)[source]

Bases: b3j0f.annotation.core.Annotation

Transforms cls into a singleton.

Reference to cls, or to any instance is the same reference.

class b3j0f.annotation.MethodMixin(function, *args, **kwargs)[source]

Bases: b3j0f.annotation.core.Annotation

Apply a mixin on a method.