Source code for autowire.implementation
from __future__ import annotations
import abc
import contextlib
from typing import Any, Callable, ContextManager, Dict, Generic, Tuple, TypeVar
from autowire.base_resource import BaseResource
from autowire.provider import ResourceProvider
R = TypeVar("R")
[docs]class Implementation(Generic[R], metaclass=abc.ABCMeta):
"""
Declarative base type for implementation
"""
[docs] @abc.abstractmethod
def reify(
self,
resource: BaseResource[R],
provider: ResourceProvider,
) -> ContextManager[R]: # pragma: no cover
"""
Reify the resource
"""
pass
[docs]class ContextManagerImplementation(Implementation[R]):
"""
Use context manager as an implementation
"""
def __init__(
self,
manager_generator: Callable[..., ContextManager[R]],
arg_resources: Tuple[BaseResource[Any], ...],
kwarg_resources: Dict[str, BaseResource[Any]],
):
super().__init__()
self.manager_generator = manager_generator
self.arg_resources = arg_resources
self.kwarg_resources = kwarg_resources
[docs] def reify(
self,
resource: BaseResource[R],
provider: ResourceProvider,
):
args = [provider.resolve(arg) for arg in self.arg_resources]
kwargs = {
name: provider.resolve(arg)
for name, arg in self.kwarg_resources.items()
}
return self.manager_generator(*args, **kwargs)
[docs]class PlainFunctionImplementation(ContextManagerImplementation[R]):
"""
Use plain function as an implementation
"""
def __init__(
self,
fn: Callable[..., R],
arg_resources: Tuple[BaseResource[Any], ...],
kwarg_resources: Dict[str, BaseResource[Any]],
):
@contextlib.contextmanager
def manager(*args, **kwargs):
yield fn(*args, **kwargs)
super().__init__(manager, arg_resources, kwarg_resources)
self.fn = fn
[docs]class ConstantImplementation(ContextManagerImplementation[R]):
"""
Use constant as an implementation
"""
def __init__(
self,
value: R
):
@contextlib.contextmanager
def manager():
yield value
super().__init__(manager, (), {})
self.value = value