Commit 8d74ab58 authored by Xavier Barbosa's avatar Xavier Barbosa

first commit

parents
[run]
omit = aiocompose/_version.py
### Python ###
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
env/
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
*.egg-info/
.installed.cfg
*.egg
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*,cover
# Translations
*.mo
*.pot
# Django stuff:
*.log
# Sphinx documentation
docs/_build/
# PyBuilder
target/
---
before_script:
- python -m pip install -e .
- python -m pip install -r requirements-test.txt
python3.3 tests:
script:
- py.test --cov aiocompose --cov-report term-missing tests/
tags:
- python3.3
python3.4 tests:
script:
- py.test --cov aiocompose --cov-report term-missing tests/
tags:
- python3.4
publish to pypi:
type: deploy
script:
- python -m pip install twine
- python setup.py sdist bdist_wheel
- twine upload -u $PYPI_USER -p $PYPI_PASSWORD dist/*
tags:
- python2.7
only:
- /^v[\d\.]+.*$/
allow_failure: true
Copyright © 2015, Xavier Barbosa
Permission is hereby granted, free of charge, to any person obtaining
a copy of this software and associated documentation files (the
"Software"), to deal in the Software without restriction, including
without limitation the rights to use, copy, modify, merge, publish,
distribute, sublicense, and/or sell copies of the Software, and to
permit persons to whom the Software is furnished to do so, subject to
the following conditions:
The above copyright notice and this permission notice shall be
included in all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
AIOCompose
==========
from .bases import Injector, annotate
from ._version import get_versions
__all__ = ['Injector', 'annotate']
__version__ = get_versions()['version']
del get_versions
This diff is collapsed.
import asyncio
import logging
from abc import ABCMeta
from collections import namedtuple, OrderedDict
from itertools import chain
logger = logging.getLogger(__name__)
class Factory:
def __init__(self, target):
self.target = target
def __call__(self, note, func=None):
def decorate(func):
self.target.factories[note] = asyncio.coroutine(func)
return func
if func:
return decorate(func)
return decorate
class FactoryMethod:
"""Decorator for func
"""
def __get__(self, obj, objtype):
target = obj or objtype
return Factory(target)
class DataStore:
def __init__(self, name, type):
self.name = name
self.type = type
def __get__(self, obj, objtype):
target = obj or objtype
if not hasattr(target, self.name):
setattr(target, self.name, self.type())
return getattr(target, self.name)
class Injector(metaclass=ABCMeta):
"""Collects dependencies and reads annotations to inject them.
"""
factory = FactoryMethod()
services = DataStore('_services', OrderedDict)
factories = DataStore('_factories', OrderedDict)
def __init__(self):
self.services = self.__class__.services.copy()
self.factories = self.__class__.factories.copy()
@asyncio.coroutine
def get(self, note):
if note in self.services:
return self.services[note]
for fact, args in note_loop(note):
if fact in self.factories:
instance = yield from self.factories[fact](*args)
logger.info('loaded service %s' % note)
self.services[note] = instance
return instance
raise ValueError('%r is not defined' % note)
@asyncio.coroutine
def inject(self, *args, **kwargs):
func, *args = args
if func in ANNOTATIONS:
annotated = ANNOTATIONS[func]
service_args, service_kwargs = [], {}
for note in annotated.pos_notes:
service = yield from self.get(note)
service_args.append(service)
for key, note in annotated.kw_notes.items():
service = yield from self.get(note)
service_kwargs[key] = service
service_args.extend(args)
service_kwargs.update(kwargs)
return func(*service_args, **service_kwargs)
logger.warn('%r is not annoted' % callable)
return callable(*args, **kwargs)
@staticmethod
def close_on_exit(obj):
"""Mark object should be closed on exit
"""
EXIT_OBJECTS.add(obj)
def close(self):
"""Closes mounted services
"""
flush = []
for name, service in self.services.items():
if service in EXIT_OBJECTS:
service.close()
logger.info('closed service %s', name)
flush.append(name)
for name in flush:
self.services.pop(name, None)
logger.info('flushed service %s', name)
ANNOTATIONS = {}
Annotation = namedtuple('Annotation', 'pos_notes kw_notes')
def annotate(*args, **kwargs):
def decorate(func):
ANNOTATIONS[func] = Annotation(args, kwargs)
return func
for arg in chain(args, kwargs.values()):
if not isinstance(arg, str):
raise ValueError('Notes must be strings')
return decorate
EXIT_OBJECTS = set()
def close_all():
for obj in EXIT_OBJECTS:
logger.info('close %s', obj)
obj.close()
def note_loop(note):
args = note.split(':')
results = []
fact, *args = args
results.append((fact, args))
while args:
suffix, *args = args
fact = '%s:%s' % (fact, suffix)
results.append((fact, args))
for fact, args in sorted(results, reverse=True):
yield fact, args
pytest
pytest-asyncio
pytest-cover
pytest-flake8
[versioneer]
VCS = git
style = pep440
versionfile_source = aiocompose/_version.py
tag_prefix = v
[metadata]
description-file = README.rst
[flake8]
exclude = _version.py
max-complexity = 10
#!/usr/bin/env python
from setuptools import setup, find_packages
import versioneer
setup(
name='aiocompose',
version=versioneer.get_version(),
author='Xavier Barbosa',
author_email='clint.northwood@gmail.com',
description='inject dependencies',
packages=find_packages(),
install_requires=[],
extras_require={
':python_version=="3.3"': ['asyncio'],
},
classifiers=[
"Development Status :: 5 - Production/Stable",
"Intended Audience :: Developers",
"License :: OSI Approved :: MIT License",
"Operating System :: OS Independent",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.3",
"Programming Language :: Python :: 3.4",
"Programming Language :: Python :: 3.5",
"Topic :: Software Development :: Libraries",
"Topic :: Software Development :: Libraries :: Python Modules"
],
keywords=['dependency injection', 'composing'],
url='http://lab.errorist.xyz/abc/aiocompose',
license='MIT',
cmdclass=versioneer.get_cmdclass()
)
import pytest
from aiocompose import Injector, annotate
@pytest.mark.asyncio
def test_instance_factory():
class MyInjector(Injector):
pass
services = MyInjector()
@services.factory('foo')
def foo_factory():
return 'I am foo'
@services.factory('bar')
def bar_factory():
return 'I am bar'
@services.factory('all')
def together_factory():
foo = yield from services.get('foo')
bar = yield from services.get('bar')
return [foo, bar]
@annotate('foo', 'bar')
def fun(foo, bar):
return {'foo': foo,
'bar': bar}
assert (yield from services.get('foo')) == 'I am foo'
assert (yield from services.get('bar')) == 'I am bar'
assert (yield from services.get('all')) == ['I am foo', 'I am bar']
assert (yield from services.inject(fun)) == {'foo': 'I am foo',
'bar': 'I am bar'}
@pytest.mark.asyncio
def test_class_factory():
class MyInjector(Injector):
pass
@MyInjector.factory('foo')
def foo_factory():
return 'I am foo'
@MyInjector.factory('bar')
def bar_factory():
return 'I am bar'
@MyInjector.factory('all')
def together_factory():
foo = yield from services.get('foo')
bar = yield from services.get('bar')
return [foo, bar]
@annotate('foo', 'bar')
def fun(foo, bar):
return {'foo': foo,
'bar': bar}
services = MyInjector()
assert (yield from services.get('foo')) == 'I am foo'
assert (yield from services.get('bar')) == 'I am bar'
assert (yield from services.get('all')) == ['I am foo', 'I am bar']
assert (yield from services.inject(fun)) == {'foo': 'I am foo',
'bar': 'I am bar'}
@pytest.mark.asyncio
def test_undefined_service_error():
class MyInjector(Injector):
pass
services = MyInjector()
with pytest.raises(ValueError):
yield from services.get('foo')
def test_annotate_error():
for val in (True, False, None, []):
with pytest.raises(ValueError):
annotate(val)
@pytest.mark.asyncio
def test_kw_inject():
class MyInjector(Injector):
pass
services = MyInjector()
@services.factory('bar')
def factory_bar():
return 'I am bar'
@annotate(foo='bar')
def fun(foo, *, bar):
return '%s, not %s' % (foo, bar)
assert (yield from services.inject(fun, bar='baz')) == 'I am bar, not baz'
@pytest.mark.asyncio
def test_late_register():
class MyInjector(Injector):
pass
def factory_foo():
return 'I am foo'
services = MyInjector()
services.factory('foo', factory_foo)
assert (yield from services.get('foo')) == 'I am foo'
@pytest.mark.asyncio
def test_sub_factory():
class MyInjector(Injector):
pass
services = MyInjector()
@services.factory('foo')
def foo_factory():
return 'I am foo'
@services.factory('foo:bar')
def bar_factory():
return 'I am bar'
assert (yield from services.get('foo')) == 'I am foo'
assert (yield from services.get('foo:bar')) == 'I am bar'
This diff is collapsed.
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment