Dependency Injection example

gistfile1.txt

from inspect import getattr_static


class dep:

    name = None

    def __init__(self, type):
        self.type = type

    def __get__(self, instance, owner):
        if self.name is None:
            raise RuntimeError("component was not initialized, use @component "
                               "decorator for {!r}".format(owner))
        if instance is None:
            return self
        else:
            try:
                return instance.__dict__[self.name]
            except KeyError:
                raise AttributeError("dep {!r} [{}] is not initialized".format(
                    self.name, self.type))

    def __set__(self, instance, value):
        raise TypeError("Cannot overwrite dependency")

    def __delete__(self, instance):
        raise TypeError("Cannot delete dependency")

    def __repr__(self):
        return '<Dep {!r} [{}]>'.format(self.name, self.type)


def component(cls):
    deps = []
    for k in dir(cls):
        v = getattr_static(cls, k)
        if isinstance(v, dep):
            # assert v.name is None, v.name
            v.name = k
            deps.append(v)
    cls.__di_deps__ = tuple(deps)
    return cls


class DI:
    """Base class for components with dependencies.

    Usage example:

        >>> @component
        ... class A:
        ...     a = dep(int)
        ...     b = dep(str)
        ...
        >>> DI(A).deps
        (<Dep 'a' [<class 'int'>]>, <Dep 'b' [<class 'str'>]>)
        >>> a = A()
        >>> DI(a).setup(a=1, b='val')
        >>> a.a
        1
        >>> a.b
        'val'
    """

    def __init__(self, inst):
        self._inst = inst
        self._deps = getattr(inst, '__di_deps__', None)
        if self._deps is None:
            raise TypeError("Argument is not component")

    @property
    def deps(self):
        """Dependencies for component"""
        return self._deps

    def setup(self, **kwargs):
        """Initializes self dependencies.

        **kwargs -- list of name->value pairs for dependencies to set up.
                    name is a name of dependency and value is a assigned value.

            >>> @component
            >>> class A:
            ...     a = dep(int)
            ...
            >>> a = A()
            >>> DI(a).setup(a=1)
            >>> a.a
            1

        Raises TypeError if value has a different type
               than has been specified in dep(type) call.
        Raises AttributeError if name doesn't point to any dependency.
        """
        marker = object()
        to_set = {}
        for d in self._deps:
            v = kwargs.pop(d.name, marker)
            if v is not marker:
                if not isinstance(v, d.type):
                    raise TypeError("Invalid type for {}, got {}"
                                    .format(d, type(v)))
                else:
                    to_set[d.name] = v
        if kwargs:
            raise AttributeError("dependencies {} are not found"
                                 .format(', '.join(sorted(kwargs))))

        # Assign new values only after all checks passed
        self._inst.__dict__.update(to_set)

    def inject(self, component, *, ready=True):
        """Injects self dependencies into component.

        component -- a component for injecting self dependencies into.
        ready -- do self.ready() check before injecting (yes by default).
        """
        cdi = DI(component)
        self_dct = self._inst.__dict__
        cdi_dct = cdi._inst.__dict__
        if ready:
            self.ready()
        marker = object()
        vals = {d.name: self_dct[d.name] for d in self._deps
                if d.name in self_dct}
        to_set = {}
        for d in cdi._deps:
            name = d.name
            selfval = vals.get(name, marker)
            if selfval is marker:
                continue  # nothing to assign
            otherval = cdi_dct.get(name, marker)
            if otherval is marker:
                if not isinstance(selfval, d.type):
                    raise TypeError("Invalid type for {}: {!r}"
                                    .format(d, selfval))
                else:
                    to_set[d.name] = selfval
                    continue
            else:
                if selfval is not otherval:
                    raise TypeError("Overrriding {} with {!r}"
                                    .format(d, selfval))
        for k, v in to_set.items():
            cdi_dct[k] = v
        return component

    def ready(self):
        """Check self for all dependencies has been initialized."""
        missed = {d.name for d in self._deps
                  if d.name not in self._inst.__dict__}
        if missed:
            raise ValueError("Not-initialized attributes {}"
                             .format(', '.join(sorted(missed))))