-
-
Save shvechikov/c28a08db6aac6f65a72e to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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)))) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment