PK!pH H component_injector/__init__.pyimport contextvars import functools import inspect from types import TracebackType from typing import Any, Callable, Dict, Optional, Type, TypeVar, cast T = TypeVar("T") TypeMap = Dict[Type[T], T] UNSET = object() class Injector: class Context: def __init__(self, injector: "Injector") -> None: self.injector = injector self.registry = injector.registry def __enter__(self) -> None: self.token = self.registry.set(self.registry.get().copy()) def __exit__( self, exc_type: Optional[Type[BaseException]], exc_val: Optional[Exception], traceback: Optional[TracebackType], ) -> None: self.registry.reset(self.token) registry: contextvars.ContextVar def __init__(self) -> None: self.registry = contextvars.ContextVar("registry", default={}) def register( self, component: Any, *, bases: bool = True, overwrite_bases: bool = True ) -> None: registry: TypeMap = self.registry.get() type_ = type(component) registry[type_] = component if bases: types = type_.mro() for type_ in types: apply = overwrite_bases or type_ not in registry if inspect.isclass(type_) and apply: registry[type_] = component def get_component(self, type_: Type[T]) -> T: registry: TypeMap = self.registry.get() return cast(T, registry[type_]) def scope(self) -> "Injector.Context": return self.Context(self) def inject(self, f: Callable[..., T]) -> Callable[..., T]: sig = inspect.signature(f) @functools.wraps(f) def wrapper(*args: Any, **kwargs: Any) -> T: registry = self.registry.get() bound = sig.bind_partial(*args, **kwargs) for name, param in sig.parameters.items(): if ( param in bound.arguments or param.annotation is inspect.Parameter.empty ): continue component = registry.get(param.annotation, UNSET) if component is not UNSET: bound.arguments[name] = component bound.apply_defaults() return f(*bound.args, **bound.kwargs) return wrapper PK!HڽTU(component_injector-1.0.0.dist-info/WHEEL A н#Z;/"d&F[xzw@Zpy3Fv]\fi4WZ^EgM_-]#0(q7PK!HG +component_injector-1.0.0.dist-info/METADATAVn6S,;[n4hh t 1-gTI*{=ɾCɲf݊̀I<9|d㟔ښ)=NwTSJmYYLkJuzU)u1ԋ%qq>)?>AM9>Av'^/M򔎫[;|o靖;t,Yҙ4I|뉟V>y϶)>pUMP­tG'Ã.7m鐾7^񇋳)]z ׋KSWM"rXnٵb>\{*܁ԙ$iںX.A;05?y*JRckSSTO\ !,*]q- L[%*A7?QEiKp:eJ1pb WByiT:աh(^ QHR[ ɚwE#S2kvA(η9m ȧ:N,ӌX  !H2:x˜G7wXk`TAE?h-4W )䲷G'zuC#sM;1A4վ+2]U@7iU8\W3c/ D]xu,B V%+fdQn\Xiq7.Hܶ`(D2 ¯³/z"S -n,6ȐGn;]# p)px~0D{oQۡe윈IwԎjzъ߼Fedo{٩%Xq̀'U`z'19Igb(G\Gnf&7=/խ*lǦG WBc~6~t`o`וU HDՀj;vS>Rp 1ؿ #*4. KyDYw'3G_D΂BTN ^ucGxhnh^`, ciURlV1( }Qȕ"uԼwdӦ(D^l2fuu NV}_vot`/ă=Q߉-X;WB?E`H0BCysF}\$_ʗХR4zwj6s2v6-蘅 "ݲ0m B