PK! Ediecast/__init__.py# -*- coding: utf-8 -*- from diecast import component # noqa from diecast import inject # noqa from diecast import registry # noqa from diecast import types # noqa PK!N4Sdiecast/component.py# -*- coding: utf-8 -*- import abc from typing import Type class Component(abc.ABC): ''' Abstract base class defining a barebones implementation of a Component type. ''' @classmethod @abc.abstractmethod def init(cls: Type['Component'], *args, **kw) -> 'Component': ''' Component initializer -- should be implemented by Component inheritors ''' pass PK!͕YYdiecast/inject.py# -*- coding: utf-8 -*- import inspect import logging from functools import wraps from inspect import Parameter from typing import Any, Callable, List, Mapping, Type, get_type_hints from diecast.types import Injector _log = logging.getLogger(__name__) def _not_injectable(registry: "ComponentRegistry", hint: Type) -> bool: return hint == Any or hint not in registry or hint is Parameter.empty def build_arg_mapping(fn: Callable) -> Mapping[str, Any]: """ Builds a dictionary mapping of argument names to the corresponding type annotations from the function. Only returns a mapping of function arguments, eliding over the return type annotation provided by `get_type_hints`. Updates the type hint for arguments without type annotations to be `typing.Any`. """ hints = get_type_hints(fn) if "return" in hints: hints.pop("return") arg_names = fn.__code__.co_varnames[: fn.__code__.co_argcount] for arg in arg_names: if arg not in hints: hints.update({arg: Any}) return hints def build_passthru_args(registry: "ComponentRegistry", fn: Callable) -> List[str]: """ Builds and returns a list of arguments (their names) that will be passed through when the injected function `fn` is called. """ args = [] sig = inspect.signature(fn) fn_params = {parm.name: parm.annotation for name, parm in sig.parameters.items()} for arg, hint in fn_params.items(): if _not_injectable(registry, hint): # This is not an injectable argument args.append(arg) return args def map_passthru_args(passthru_args: List[str], *args, **kw) -> Mapping[str, Any]: """ Builds a mapping of passthrough args to their values. Only maps passed through args into `**kw` that will be passed into the function. """ arg_map = {} # If `*args` or `*kw` contain an ellipses, drop that entry args = list(filter(lambda item: item is not ..., args)) kw = dict(filter(lambda item: item[1] is not ..., kw.items())) # Apply *args in order for name, val in zip(passthru_args, args): arg_map.update({name: val}) for name, val in kw.items(): if name in passthru_args: if name in arg_map: raise ValueError(f"Passthru arg {name} mapped through *args and **kw") arg_map.update({name: val}) return arg_map def make_injector(registry: "ComponentRegistry") -> Injector: """ This is just a "magical" function that returns a decorator that has been bound to a registry. `registry` is closured in to `_arg_injector`. """ def _injector(fn: Callable): @wraps(fn) def _arg_injector(*args, **kw): sig_params = _do_inject(registry, fn, *args, **kw) return _call_with_bound_params(fn, sig_params) return _arg_injector return _injector def _do_inject( _registry: "ComponentRegistry", _fn: Callable, *args, **kw ) -> inspect.BoundArguments: """ Given a `_registry`, `_fn` to inject, as well as the arguments to be passed to `_fn`, creates an `inspect.Signature` for the function. We then bind all injected and passthrough arguments to the `inspect.Signature` using `sig.bind_partial`. Funnily enough, this does not *actually* bind the function itself, or even execute the function. """ _log.debug(f"Performing injection for {_fn} with {_registry}") sig = inspect.signature(_fn) arg_map = build_arg_mapping(_fn) injected_params = {} for arg, hint in arg_map.items(): if _not_injectable(_registry, hint): continue dep = _registry.get(hint) if dep["instance"] is None: # This branch happens on non-persisted components _log.debug(f"Initialize dependency for injection {dep}") # registry[item] is a shortcut for initializing a component # from a hint injected_params.update({arg: _registry[hint]}) else: # This branch happens on persisted components injected_params.update({arg: dep.get("instance")}) passthru_args = build_passthru_args(_registry, _fn) passthru_params = map_passthru_args(passthru_args, *args, **kw) params = dict(injected_params) params.update(passthru_params) return sig.bind_partial(**params) def _call_with_bound_params(fn: Callable, sig_params: inspect.BoundArguments) -> Any: """ Uses a Signature's BoundArguments to execute the underlying function. """ return fn(*sig_params.args, **sig_params.kwargs) PK!&> diecast/registry.py# -*- coding: utf-8 -*- from typing import ( Callable, Dict, Optional, Type, ) from diecast.component import Component from diecast.inject import _call_with_bound_params, _do_inject from diecast.types import ComponentState class ComponentRegistry(object): def __init__(self): self._components: Dict[Type[Component], ComponentState] = {} def __getitem__(self, cls: Type[Component]) -> Component: ''' Returns a component instance or creates a new component instance. ''' component = self.get(cls) instance = component.get('instance') if not instance: return self._init_component(component.get('init')) else: return instance def __contains__(self, cls: Type[Component]) -> bool: ''' Returns whether or not this registry contains component `cls`. ''' return cls in self._components def _init_component(self, init: Callable[..., Component]) -> Component: if not init: return None sig_params = _do_inject(self, init) return _call_with_bound_params(init, sig_params) def get(self, cls: Type[Component]) -> ComponentState: ''' Returns ComponentState for a Component. ''' return self._components.get(cls) def add(self, cls: Type[Component], init: Optional[Callable[..., Component]]=None, persist: bool=True): ''' Add a component to this component registry. `cls` is the class that will be registered in the registry. `init` is the Callable that will provide an instance of `cls` for injection. Before `init` is called, dependency injection is performed on the function. `persist` specifies that the created instance should be stored in the registry. ''' if not init: init = cls.init instance = None if persist: # This should perform DI on the `init` callable to # instantiate the Component instance = self._init_component(init) self._components.update({ cls: ComponentState({ 'init': init, 'persist': persist, 'instance': instance, }), }) __components: ComponentRegistry = ComponentRegistry() def get_registry() -> ComponentRegistry: ''' Returns the default global registry. ''' global __components return __components def register_component(cls: Type[Component], init: Optional[Callable[..., Component]]=None, persist: bool=True, registry: ComponentRegistry=None): ''' Register a component with the dependency injection system. `cls` is the class that will be registered in the registry. `init` is the Callable that will provide an instance of `cls` for injection. Before `init` is called, dependency injection is performed on the function. `persist` specifies that the created instance should be stored in the registry. `registry` is the `ComponentRegistry` instance that the component will be registered with. If `None`, will use the default global registry. ''' if registry is None: global __components registry = __components assert cls not in registry return registry.add( cls=cls, init=init, persist=persist, ) PK! 5diecast/types.py# -*- coding: utf-8 -*- import inspect from typing import ( Any, Callable, Dict, NewType, ) from diecast.component import Component ComponentState = NewType('ComponentState', Dict[str, Any]) Injector = NewType( 'Injector', Callable[[Callable], Callable], ) PK!2  diecast-0.3.2.dist-info/LICENSECopyright (c) 2018 Sean Johnson 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. PK!HnHTUdiecast-0.3.2.dist-info/WHEEL A н#Z;/"d&F[xzw@Zpy3Fv]\fi4WZ^EgM_-]#0(q7PK!H@% diecast-0.3.2.dist-info/METADATAQN1)vȢ& DA f;,Ecۅ;BqO|cJ&aYC~ޤAJc%cbр|V12УUhvU"q.|XiC͒=;5nRQEspύԎ,mlt8t6e>`ؒotJ6i V(-4fdi;Ο-c~J-l<1Fb z,i1Ff mCUG׽7Pǣ˙%zurzPK!H韲diecast-0.3.2.dist-info/RECORDuɒ0{? ,IDmE rI aw:3 |W%q aZ! QM`ZOeUk)SoA~ͭU6Td ||1VlyC՘UWC|V:Xi M_{$ o|r6y-M7 bohx2^ 6diecast/registry.pyPK! 5>#diecast/types.pyPK!2  $diecast-0.3.2.dist-info/LICENSEPK!HnHTU(diecast-0.3.2.dist-info/WHEELPK!H@% t)diecast-0.3.2.dist-info/METADATAPK!H韲*diecast-0.3.2.dist-info/RECORDPK s,