Last active
October 24, 2016 06:03
-
-
Save fre-sch/2f60b289afae39fcf1ce71313d40b3c4 to your computer and use it in GitHub Desktop.
IOC and DI service locator container implementation
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 sslib import Registry | |
import sys | |
sys.modules[__name__] = Registry() |
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
import components | |
# register a factory function using the factory decorator | |
@components.factory("config") | |
def load_configuration(): | |
return json.load("config.json") | |
# register a factory class using the factory decorator | |
@components.factory("user_service") | |
class UserService: | |
def find_email(self, email): | |
return {"name": "test", "email": email} | |
# inject and use components in a function | |
@components.inject("config") | |
@components.inject("user_service") | |
def main(config=None, user_service=None) | |
print("configuration: {!r}".format(config)) | |
print("user: {!r}".format(user_service.find_email("[email protected]")) |
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
# coding: utf-8 | |
import logging | |
from functools import wraps, partial | |
from pprint import pformat | |
log = logging.getLogger(__name__) | |
class SllibError(Exception): | |
pass | |
class MissingComponentError(SllibError): | |
def __init__(self, name): | |
super(MissingComponentError, self).__init__( | |
"missing component factory for {!r}".format(name)) | |
class Wire: | |
""" | |
Non-data descriptor resolving component | |
""" | |
def __init__(self, registry, name): | |
self.registry = registry | |
self.name = name | |
def __get__(self, obj, cls): | |
if obj is None: | |
return self | |
instance = getattr(self.registry, self.name) | |
return instance | |
def injector(registry, names, fn): | |
@wraps(fn) | |
def inner(*args, **kwargs): | |
if not any(name in kwargs for name in names): | |
return fn(*args, **kwargs) | |
new_kwargs = kwargs.copy() | |
for name in names: | |
if name not in kwargs: | |
new_kwargs[name] = getattr(registry, name) | |
return fn(*args, **new_kwargs) | |
return inner | |
class Container: | |
""" | |
Service locator container. | |
Create singleton container for project: | |
>>> from sllib import Container | |
>>> import sys | |
>>> instance = Container() | |
>>> instance.__file__ = __file__ | |
>>> sys.modules[__name__] = instance | |
And inject some components into a function/method: | |
>>> import components | |
>>> @components.inject("config", "db") | |
>>> def user_service(config=None, db=None): | |
... pass | |
Or wire some components as properties: | |
>>> import components | |
>>> class Example: | |
... db_service = components.wire("db_service") | |
... def find_all(self): | |
... self.db_service.find_all() | |
""" | |
def __init__(self): | |
self.factories = {} | |
def factory(self, name, factory=None): | |
""" | |
Add factory for name. | |
Use without factory param as decorator, or with factory param to add | |
factory directly. | |
:param name: string name of service | |
:param factory: callable, optional | |
""" | |
if factory is None: | |
def inner(factory): | |
self.factories[name] = factory | |
return factory | |
return inner | |
else: | |
self.factories[name] = factory | |
def __getattr__(self, name): | |
""" | |
Get instance of component or construct new instance from factory. | |
""" | |
if name not in self.factories: | |
raise MissingComponentError(name) | |
factory = self.factories[name] | |
instance = factory() | |
log.info("initialize %r from %r", name, factory) | |
setattr(self, name, instance) | |
return instance | |
def wire(self, name): | |
""" | |
Wire some component as property | |
>>> class Example: | |
... db_service = components.wire("db_service") | |
""" | |
return Wire(self, name) | |
def inject(self, *names): | |
""" | |
Inject some components as params | |
>>> @components.inject("config", "db") | |
>>> def example(config=None, db=None): | |
... pass | |
""" | |
return partial(injector, self, names) | |
def __repr__(self): | |
return "Registry({})".format(pformat(self.__dict__)) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment