jacked._inject._check_decorated() - Code Metrics - ramonhagenaars/jacked - Measure and Improve Code Quality continuously with Scrutinizer

jacked._inject._check_decorated() A
last analyzed 2019-06-21 12:21 UTC

↳ Parent: jacked._inject

Complexity

Conditions

Size

Total Lines	11
Code Lines	8

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	4
eloc	8
nop	1
dl	0
loc	11
rs	10
c	0
b	0
f	0

"""
PRIVATE MODULE: do not import (from) it directly.

This module contains the ``inject`` function and its required private
functions.
"""
import functools
import inspect
from collections import ChainMap
from functools import partial, lru_cache
from pathlib import Path
from typing import List, Dict, Any, Type, Tuple
from jacked import _container
from jacked._container import DEFAULT_CONTAINER
from jacked._discover import discover
from jacked._exceptions import InjectionError, InvalidUsageError
from jacked._injectable import Injectable
from jacked._typing import T
from jacked.matchers._base_matcher import BaseMatcher


def inject_here(
        hint: Type[T],
        *,
        container: _container.Container = _container.DEFAULT_CONTAINER
) -> T:
    """
    Usage example:

        some_inst = inject_here(SomeClass)

    :param hint: the type that hints what is to be returned.
    :param container: the Container from which the injectable is to be
    returned.
    :return: an injectable that corresponds to ``hint``.
    """
    candidates = _get_candidates(hint, container)
    if not candidates:
        raise InjectionError('No suitable candidates for "{}".'
                             .format(hint), hint)
    return _choose_candidate(candidates)


def inject(
        decorated: callable = None,
        *,
        container: _container.Container = _container.DEFAULT_CONTAINER
) -> callable:
    """
    Decorator that will inject all parameters that were not already explicitly
    provided.

    Usage example:

        @inject
        def func(x: SomeClass):
            x.some_func()  # x is now an instance of SomeClass.

    :param decorated: the callable that is decorated.
    :param container: the storage that is used that contains all
    ``Injectables``.
    :return: a decorator.
    """
    if decorated:
        _check_decorated(decorated)
        return functools.update_wrapper(
            lambda *args, **kwargs: _wrapper(decorated, container,
                                             *args, **kwargs), decorated)
    return partial(_decorator, container=container)


def get_candidates(
        hint: T,
        *,
        container: _container.Container = DEFAULT_CONTAINER) -> List[T]:
    """
    Return all candidates for the given type ``T`` and return them in a list.
    :param hint: the type for which candidates are to be returned.
    :param container: the container from which the injectables are fetched.
    :return: a list of candidates of type ``T``.
    """
    return [c for c, _ in _get_candidates(hint, container)]


def _decorator(
        decorated: callable,
        container: _container.Container) -> callable:
    # This function acts as the "actual decorator" if any arguments were passed
    # to `inject`.
    _check_decorated(decorated)
    return functools.update_wrapper(
        lambda *args, **kwargs: _wrapper(decorated, container, *args,
                                         **kwargs), decorated)


def _check_decorated(decorated: callable):
    # This function validates the decorated object and raises upon an invalid
    # decoration.
    if isinstance(decorated, type):
        raise InvalidUsageError('The inject decorator can be used on '
                                'callables only.')
    params = inspect.signature(decorated).parameters
    for param_name in params:
        hint = params[param_name].annotation
        if hint != inspect.Parameter.empty:
            _check_hint(hint)


def _check_hint(hint: Any):
    pass  # TODO implement this.


def _wrapper(
        decorated: callable,
        container: _container.Container,
        *args,
        **kwargs_):
    # This function is wrapped around the decorated object. It will collect
    # arguments and inject them to `decorated` by providing these arguments.
    signature = inspect.signature(decorated)

    if args:
        # If there are any ordered parameters given, filter them out of the
        # signature to prevent the search for injection candidates:
        filtered_params = list(signature.parameters.values())[len(args):]
        signature = inspect.Signature(filtered_params)

    # Collect the arguments for injection:
    arguments = _collect_arguments(signature, container)
    kwargs_ = ChainMap(kwargs_, arguments)  # Note: kwargs_ takes precedence.

    # Now all arguments are collected, "inject" them into `decorated`:
    return decorated(*args, **kwargs_)


def _collect_arguments(
        signature: inspect.Signature,
        container: _container.Container) -> Dict[str, object]:
    # This function tries to collect arguments for the given signature and
    # returns a dictionary that corresponds to that signature.
    result = {}
    for param_name in signature.parameters:
        if param_name in ('self', 'cls'):
            continue
        param = signature.parameters[param_name]
        hint = param.annotation
        # Get all candidates that could be injected according to `signature`:
        candidates = _get_candidates(hint, container)
        if not candidates:
            result[param_name] = param.default
            if param.default is inspect.Parameter.empty:
                raise InjectionError('No suitable candidates for "{}".'
                                     .format(param_name), param)
        else:
            # If there are multiple candidates, select one:
            result[param_name] = _choose_candidate(candidates)
    return result


def _get_candidates(
        hint: T,
        container: _container.Container) -> List[Tuple[T, Injectable]]:
    # Search in the known injectables in `container` for all matching
    # candidates. The candidates are returned sorted by their priority.
    candidates = ((_match(hint, injectable, container), injectable)

                  for injectable in container.injectables)
    result = [(c, i) for c, i in candidates if c]
    result.sort(key=lambda c: c[1].priority, reverse=True)
    return result


def _choose_candidate(candidates: List[Tuple[T, Injectable]]) -> T:
    # From a list of candidates, pick and return one:
    return candidates[0][0]  # The first should have the highest priority.


def _match(
        hint: type,
        injectable: Injectable,
        container: _container.Container) -> object:
    # Check if the given `parameter` matches with the given `injectable`. If
    # there appears to be a match, return what is to be injected (e.g. an
    # instance of a class, a class itself, ...). If no match, return `None`.
    for matcher in _get_matchers():
        if matcher.can_match(hint):
            # Match or no match, return anyway:
            return matcher.match(hint, injectable, container)


@lru_cache()
def _get_matchers() -> List[BaseMatcher]:
    path_to_matchers = str(Path(__file__).parent.joinpath(Path('matchers')))
    modules = discover(path_to_matchers)

    public_elements = [getattr(mod, elem) for mod in modules
                       for elem in dir(mod) if not elem.startswith('_')]
    matchers = [cls() for cls in public_elements if isinstance(cls, type)
                and cls is not BaseMatcher and issubclass(cls, BaseMatcher)]

    matchers.sort(key=lambda m: m.priority(), reverse=True)

    return matchers


1			"""
2			PRIVATE MODULE: do not import (from) it directly.
3
4			This module contains the ``inject`` function and its required private
5			functions.
6			"""
7			import functools
8			import inspect
9			from collections import ChainMap
10			from functools import partial, lru_cache
11			from pathlib import Path
12			from typing import List, Dict, Any, Type, Tuple
13			from jacked import _container
14			from jacked._container import DEFAULT_CONTAINER
15			from jacked._discover import discover
16			from jacked._exceptions import InjectionError, InvalidUsageError
17			from jacked._injectable import Injectable
18			from jacked._typing import T
19			from jacked.matchers._base_matcher import BaseMatcher
20
21
22			def inject_here(
23			hint: Type[T],
24			*,
25			container: _container.Container = _container.DEFAULT_CONTAINER
26			) -> T:
27			"""
28			Usage example:
29
30			some_inst = inject_here(SomeClass)
31
32			:param hint: the type that hints what is to be returned.
33			:param container: the Container from which the injectable is to be
34			returned.
35			:return: an injectable that corresponds to ``hint``.
36			"""
37			candidates = _get_candidates(hint, container)
38			if not candidates:
39			raise InjectionError('No suitable candidates for "{}".'
40			.format(hint), hint)
41			return _choose_candidate(candidates)
42
43
44			def inject(
45			decorated: callable = None,
46			*,
47			container: _container.Container = _container.DEFAULT_CONTAINER
48			) -> callable:
49			"""
50			Decorator that will inject all parameters that were not already explicitly
51			provided.
52
53			Usage example:
54
55			@inject
56			def func(x: SomeClass):
57			x.some_func() # x is now an instance of SomeClass.
58
59			:param decorated: the callable that is decorated.
60			:param container: the storage that is used that contains all
61			``Injectables``.
62			:return: a decorator.
63			"""
64			if decorated:
65			_check_decorated(decorated)
66			return functools.update_wrapper(
67			lambda args, *kwargs: _wrapper(decorated, container,
68			args, *kwargs), decorated)
69			return partial(_decorator, container=container)
70
71
72			def get_candidates(
73			hint: T,
74			*,
75			container: _container.Container = DEFAULT_CONTAINER) -> List[T]:
76			"""
77			Return all candidates for the given type ``T`` and return them in a list.
78			:param hint: the type for which candidates are to be returned.
79			:param container: the container from which the injectables are fetched.
80			:return: a list of candidates of type ``T``.
81			"""
82			return [c for c, _ in _get_candidates(hint, container)]
83
84
85			def _decorator(
86			decorated: callable,
87			container: _container.Container) -> callable:
88			# This function acts as the "actual decorator" if any arguments were passed
89			# to `inject`.
90			_check_decorated(decorated)
91			return functools.update_wrapper(
92			lambda args, kwargs: _wrapper(decorated, container, args,
93			**kwargs), decorated)
94
95
96			def _check_decorated(decorated: callable):
97			# This function validates the decorated object and raises upon an invalid
98			# decoration.
99			if isinstance(decorated, type):
100			raise InvalidUsageError('The inject decorator can be used on '
101			'callables only.')
102			params = inspect.signature(decorated).parameters
103			for param_name in params:
104			hint = params[param_name].annotation
105			if hint != inspect.Parameter.empty:
106			_check_hint(hint)
107
108
109			def _check_hint(hint: Any):
110			pass # TODO implement this.
111
112
113			def _wrapper(
114			decorated: callable,
115			container: _container.Container,
116			*args,
117			**kwargs_):
118			# This function is wrapped around the decorated object. It will collect
119			# arguments and inject them to `decorated` by providing these arguments.
120			signature = inspect.signature(decorated)
121
122			if args:
123			# If there are any ordered parameters given, filter them out of the
124			# signature to prevent the search for injection candidates:
125			filtered_params = list(signature.parameters.values())[len(args):]
126			signature = inspect.Signature(filtered_params)
127
128			# Collect the arguments for injection:
129			arguments = _collect_arguments(signature, container)
130			kwargs_ = ChainMap(kwargs_, arguments) # Note: kwargs_ takes precedence.
131
132			# Now all arguments are collected, "inject" them into `decorated`:
133			return decorated(args, *kwargs_)
134
135
136			def _collect_arguments(
137			signature: inspect.Signature,
138			container: _container.Container) -> Dict[str, object]:
139			# This function tries to collect arguments for the given signature and
140			# returns a dictionary that corresponds to that signature.
141			result = {}
142			for param_name in signature.parameters:
143			if param_name in ('self', 'cls'):
144			continue
145			param = signature.parameters[param_name]
146			hint = param.annotation
147			# Get all candidates that could be injected according to `signature`:
148			candidates = _get_candidates(hint, container)
149			if not candidates:
150			result[param_name] = param.default
151			if param.default is inspect.Parameter.empty:
152			raise InjectionError('No suitable candidates for "{}".'
153			.format(param_name), param)
154			else:
155			# If there are multiple candidates, select one:
156			result[param_name] = _choose_candidate(candidates)
157			return result
158
159
160			def _get_candidates(
161			hint: T,
162			container: _container.Container) -> List[Tuple[T, Injectable]]:
163			# Search in the known injectables in `container` for all matching
164			# candidates. The candidates are returned sorted by their priority.
165			candidates = ((_match(hint, injectable, container), injectable)
			0 ignored issues – show Comprehensibility Best Practice introduced 2019-06-09 09:50 UTC by Report Bug Copy Issue Report The variable `injectable` does not seem to be defined. Loading history...
166			for injectable in container.injectables)
167			result = [(c, i) for c, i in candidates if c]
168			result.sort(key=lambda c: c[1].priority, reverse=True)
169			return result
170
171
172			def _choose_candidate(candidates: List[Tuple[T, Injectable]]) -> T:
173			# From a list of candidates, pick and return one:
174			return candidates[0][0] # The first should have the highest priority.
175
176
177			def _match(
178			hint: type,
179			injectable: Injectable,
180			container: _container.Container) -> object:
181			# Check if the given `parameter` matches with the given `injectable`. If
182			# there appears to be a match, return what is to be injected (e.g. an
183			# instance of a class, a class itself, ...). If no match, return `None`.
184			for matcher in _get_matchers():
185			if matcher.can_match(hint):
186			# Match or no match, return anyway:
187			return matcher.match(hint, injectable, container)
188
189
190			@lru_cache()
191			def _get_matchers() -> List[BaseMatcher]:
192			path_to_matchers = str(Path(__file__).parent.joinpath(Path('matchers')))
193			modules = discover(path_to_matchers)
194
195			public_elements = [getattr(mod, elem) for mod in modules
196			for elem in dir(mod) if not elem.startswith('_')]
197			matchers = [cls() for cls in public_elements if isinstance(cls, type)
198			and cls is not BaseMatcher and issubclass(cls, BaseMatcher)]
199
200			matchers.sort(key=lambda m: m.priority(), reverse=True)
201
202			return matchers
203

ramonhagenaars / jacked

jacked._inject._check_decorated() A last analyzed 2019-06-21 12:21 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

jacked._inject._check_decorated() A
last analyzed 2019-06-21 12:21 UTC