pyjection.dependency_injector.DependencyInjector._generate_arguments_dict() - Code Metrics - Inspection of "Enable code intelligence in scrutinizer" - Darkheir/pyjection - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Branch — master (62458d)

by Raphaël

created 2018-01-04 09:17 UTC

DependencyInjector._generate_arguments_dict() B

↳ Parent: pyjection.dependency_injector

Complexity

Conditions

Size

Total Lines	31
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	10
CRAP Score	4.074

Importance

Changes

Metric	Value
eloc	12
dl	0
loc	31
rs	8.5806
c	0
b	0
f	0
ccs	10
cts	12
cp	0.8333
cc	4
nop	2
crap	4.074

import logging

from inspect import signature
from inspect import Parameter
from collections import OrderedDict

from pyjection.resolvers import ServiceResolver, NameResolver
from pyjection.service import Service
from pyjection.helper import get_service_subject_identifier


class DependencyInjector(object):
    """
    Main class of the package.

    This is the interface that should be used to get objects from the dependency injector.
    """

    def __init__(self, resolvers=None):
        self._logger = logging.getLogger(__name__)
        self._services = dict()
        self._singletons = dict()
        self._resolvers = resolvers
        if not resolvers:
            self._resolvers = [
                ServiceResolver(self),
                NameResolver(self)
            ]

    def register(self, service_subject, identifier=None):
        """
        Register a new service in the dependency injector

        This service can be :
            * A class that will be instantiated when called
            * An already instantiated instance that will be returned

        If no identifier is passed, it will be the class name in snake_case

        :param service_subject: The class or instance
        :type service_subject: mixed
        :param identifier: The identifier used to later retrieve a service instance
        :type identifier: string

        :return: Return the newly created service entry
        :rtype: Service
        """
        if identifier is None:
            identifier = get_service_subject_identifier(service_subject)
        service = Service(service_subject)
        self._services[identifier] = service
        self._logger.debug(
            "Class %s registered with identifier %s",
            str(service_subject),
            identifier
        )
        return service

    def register_singleton(self, service_subject, identifier=None):
        """
        Register a new singleton service in in the dependency injector

        This service can be :
            * A class that will be instantiated when called
            * An already instantiated instance that will be returned

        If no identifier is passed, it will be the class name in snake_case

        :param service_subject: The class or instance
        :type service_subject: mixed
        :param identifier: The identifier used to later retrieve a service singleton
        :type identifier: string

        :return: Return the newly created dependency entry
        :rtype: Service
        """
        if identifier is None:
            identifier = get_service_subject_identifier(service_subject)
        service = Service(service_subject)
        service.is_singleton = True
        self._services[identifier] = service
        self._logger.debug(
            "Class %s registered as singleton with identifier %s",
            str(service_subject),
            identifier
        )
        return service

    def get(self, identifier):
        """
        Instantiate and retrieve the service matching this identifier

        If the service has been self has a singleton the same service object
        will be return each time this service is asked

        :param identifier: The identifier or the class to retrieve
        :type identifier: mixed
        :return: The instantiated object
        :rtype: mixed
        """
        if isinstance(identifier, str) is False:
            identifier = get_service_subject_identifier(identifier)

        self._validate_service_name(identifier)

        service = self._services[identifier]
        instance = self._get_singleton(identifier, service)
        if instance is not None:
            self._logger.debug("Return singleton with ID %s", identifier)
            return instance

        instance = self._get_instance(service)
        self._set_singleton(identifier, instance, service)
        self._logger.debug("Return instance with ID %s", identifier)
        return instance

    def get_uninstantiated(self, identifier):
        if isinstance(identifier, str) is False:
            identifier = get_service_subject_identifier(identifier)

        self._validate_service_name(identifier)

        service = self._services[identifier]
        return service.subject

    def has_service(self, identifier):
        """
        Check if the service matching the given identifier
        has already been declared

        :param identifier: Name of the service or the class
        :type identifier: mixed
        :return: Whether or not the service exists
        :rtype: boolean
        """
        if isinstance(identifier, str) is False:
            identifier = get_service_subject_identifier(identifier)

        if identifier in self._services:
            return True
        return False

    def _validate_service_name(self, identifier):
        if not self.has_service(identifier):
            self._logger.error("No service has been declared with ID %s", identifier)
            raise Exception("No service has been declared with this ID")

    def _get_singleton(self, identifier, service):
        """
        Return the singleton if it has been setted and
        the service represents a singleton

        :param identifier: the singleton identifier
        :param service: The service we need the singleton for
        :type identifier: string
        :type service: Service

        :return: The singleton instance or None
        :rtype: mixed
        """
        if service.is_singleton is True and identifier in self._singletons:
            return self._singletons[identifier]
        return None

    def _set_singleton(self, identifier, instance, service):
        """
        Set the instance as a singleton in the dict
        if the service represents a singleton

        :param identifier: the singleton identifier
        :param service: The service we want to set a singleton for
        :param instance: The singleton instance
        :type identifier: string
        :type service: Service
        :type instance: mixed
        """
        if service.is_singleton is True:
            self._singletons[identifier] = instance

    def _get_instance(self, service):
        """
        Return the instantiated object for the given service

        :param service: The service we need an instance for
        :type service: Service
        :return: The instantiated object
        """
        if service.type == 'instance':
            return service.subject
        arguments = self._generate_arguments_dict(service)
        return service.subject(**arguments)

    def _generate_arguments_dict(self, service):
        """
        Generate a dict containing all the parameters values
        required to Instantiate the service.

        An exception is raised if a mandatory argument cannot be
        retrieved.

        :param service: The service that needs to be instantiated
        :type service: Service
        :return: The parameters values to use to instantiate the service
        :rtype: dict
        """
        arguments = dict()

        # We can't use signature on class object __init__
        if self._is_object_init(service.subject):
            return arguments

        sig = signature(service.subject.__init__)
        method_parameters = OrderedDict(sig.parameters)

        # Pop the first param since it's the self class instance
        method_parameters.popitem(False)

        for method_parameter in method_parameters.values():
            argument = self._get_argument(service, method_parameter)
            if argument is not None:
                arguments[method_parameter.name] = argument

        return arguments

    @staticmethod
    def _is_object_init(subject):
        """
        Check if the __init__ method for the object comes from
        the default object class or has been overridden

        :param subject: The subject we want to check the __init__ for
        :type subject: mixed

        :return: Whether the __init__ method is the default on or not
        :rtype: boolean
        """
        if '__objclass__' in dir(subject.__init__) and subject.__init__.__objclass__ == object:
            return True
        return False

    def _get_argument(self, service, method_parameter):
        """
        Retrieve the argument value for the given service

        :param service: The service we need an argument for
        :param method_parameter: The parameter we need the value for
        :type service: Service
        :type method_parameter: Parameter
        :return: The argument value
        :rtype: mixed
        """
        for resolver in self._resolvers:
            resolved = resolver.resolve(method_parameter, service)
            if resolved:
                return resolved

        # If the parameter is *args or **kwargs or has a default value
        # then we don't raise any exception
        if (method_parameter.kind in [Parameter.VAR_POSITIONAL, Parameter.VAR_KEYWORD]
                or method_parameter.default is not Parameter.empty):
            return None

        self._logger.error("A required argument is not set: %s", method_parameter.name)
        raise Exception("A required argument is not set (%s)" % method_parameter.name)


1	1	import logging
2
3	1	from inspect import signature
4	1	from inspect import Parameter
5	1	from collections import OrderedDict
6
7	1	from pyjection.resolvers import ServiceResolver, NameResolver
8	1	from pyjection.service import Service
9	1	from pyjection.helper import get_service_subject_identifier
10
11
12	1	class DependencyInjector(object):
13		"""
14		Main class of the package.
15
16		This is the interface that should be used to get objects from the dependency injector.
17		"""
18
19	1	def __init__(self, resolvers=None):
20	1	self._logger = logging.getLogger(__name__)
21	1	self._services = dict()
22	1	self._singletons = dict()
23	1	self._resolvers = resolvers
24	1	if not resolvers:
25	1	self._resolvers = [
26		ServiceResolver(self),
27		NameResolver(self)
28		]
29
30	1	def register(self, service_subject, identifier=None):
31		"""
32		Register a new service in the dependency injector
33
34		This service can be :
35		* A class that will be instantiated when called
36		* An already instantiated instance that will be returned
37
38		If no identifier is passed, it will be the class name in snake_case
39
40		:param service_subject: The class or instance
41		:type service_subject: mixed
42		:param identifier: The identifier used to later retrieve a service instance
43		:type identifier: string
44
45		:return: Return the newly created service entry
46		:rtype: Service
47		"""
48	1	if identifier is None:
49	1	identifier = get_service_subject_identifier(service_subject)
50	1	service = Service(service_subject)
51	1	self._services[identifier] = service
52	1	self._logger.debug(
53		"Class %s registered with identifier %s",
54		str(service_subject),
55		identifier
56		)
57	1	return service
58
59	1	def register_singleton(self, service_subject, identifier=None):
60		"""
61		Register a new singleton service in in the dependency injector
62
63		This service can be :
64		* A class that will be instantiated when called
65		* An already instantiated instance that will be returned
66
67		If no identifier is passed, it will be the class name in snake_case
68
69		:param service_subject: The class or instance
70		:type service_subject: mixed
71		:param identifier: The identifier used to later retrieve a service singleton
72		:type identifier: string
73
74		:return: Return the newly created dependency entry
75		:rtype: Service
76		"""
77	1	if identifier is None:
78	1	identifier = get_service_subject_identifier(service_subject)
79	1	service = Service(service_subject)
80	1	service.is_singleton = True
81	1	self._services[identifier] = service
82	1	self._logger.debug(
83		"Class %s registered as singleton with identifier %s",
84		str(service_subject),
85		identifier
86		)
87	1	return service
88
89	1	def get(self, identifier):
90		"""
91		Instantiate and retrieve the service matching this identifier
92
93		If the service has been self has a singleton the same service object
94		will be return each time this service is asked
95
96		:param identifier: The identifier or the class to retrieve
97		:type identifier: mixed
98		:return: The instantiated object
99		:rtype: mixed
100		"""
101	1	if isinstance(identifier, str) is False:
102	1	identifier = get_service_subject_identifier(identifier)
103
104	1	self._validate_service_name(identifier)
105
106	1	service = self._services[identifier]
107	1	instance = self._get_singleton(identifier, service)
108	1	if instance is not None:
109	1	self._logger.debug("Return singleton with ID %s", identifier)
110	1	return instance
111
112	1	instance = self._get_instance(service)
113	1	self._set_singleton(identifier, instance, service)
114	1	self._logger.debug("Return instance with ID %s", identifier)
115	1	return instance
116
117	1	def get_uninstantiated(self, identifier):
118		if isinstance(identifier, str) is False:
119		identifier = get_service_subject_identifier(identifier)
120
121		self._validate_service_name(identifier)
122
123		service = self._services[identifier]
124		return service.subject
125
126	1	def has_service(self, identifier):
127		"""
128		Check if the service matching the given identifier
129		has already been declared
130
131		:param identifier: Name of the service or the class
132		:type identifier: mixed
133		:return: Whether or not the service exists
134		:rtype: boolean
135		"""
136	1	if isinstance(identifier, str) is False:
137	1	identifier = get_service_subject_identifier(identifier)
138
139	1	if identifier in self._services:
140	1	return True
141	1	return False
142
143	1	def _validate_service_name(self, identifier):
144	1	if not self.has_service(identifier):
145	1	self._logger.error("No service has been declared with ID %s", identifier)
146	1	raise Exception("No service has been declared with this ID")
147
148	1	def _get_singleton(self, identifier, service):
149		"""
150		Return the singleton if it has been setted and
151		the service represents a singleton
152
153		:param identifier: the singleton identifier
154		:param service: The service we need the singleton for
155		:type identifier: string
156		:type service: Service
157
158		:return: The singleton instance or None
159		:rtype: mixed
160		"""
161	1	if service.is_singleton is True and identifier in self._singletons:
162	1	return self._singletons[identifier]
163	1	return None
164
165	1	def _set_singleton(self, identifier, instance, service):
166		"""
167		Set the instance as a singleton in the dict
168		if the service represents a singleton
169
170		:param identifier: the singleton identifier
171		:param service: The service we want to set a singleton for
172		:param instance: The singleton instance
173		:type identifier: string
174		:type service: Service
175		:type instance: mixed
176		"""
177	1	if service.is_singleton is True:
178	1	self._singletons[identifier] = instance
179
180	1	def _get_instance(self, service):
181		"""
182		Return the instantiated object for the given service
183
184		:param service: The service we need an instance for
185		:type service: Service
186		:return: The instantiated object
187		"""
188	1	if service.type == 'instance':
189	1	return service.subject
190	1	arguments = self._generate_arguments_dict(service)
191	1	return service.subject(**arguments)
192
193	1	def _generate_arguments_dict(self, service):
194		"""
195		Generate a dict containing all the parameters values
196		required to Instantiate the service.
197
198		An exception is raised if a mandatory argument cannot be
199		retrieved.
200
201		:param service: The service that needs to be instantiated
202		:type service: Service
203		:return: The parameters values to use to instantiate the service
204		:rtype: dict
205		"""
206	1	arguments = dict()
207
208		# We can't use signature on class object __init__
209	1	if self._is_object_init(service.subject):
210		return arguments
211
212	1	sig = signature(service.subject.__init__)
213	1	method_parameters = OrderedDict(sig.parameters)
214
215		# Pop the first param since it's the self class instance
216	1	method_parameters.popitem(False)
217
218	1	for method_parameter in method_parameters.values():
219	1	argument = self._get_argument(service, method_parameter)
220	1	if argument is not None:
221		arguments[method_parameter.name] = argument
222
223	1	return arguments
224
225	1	@staticmethod
226		def _is_object_init(subject):
227		"""
228		Check if the __init__ method for the object comes from
229		the default object class or has been overridden
230
231		:param subject: The subject we want to check the __init__ for
232		:type subject: mixed
233
234		:return: Whether the __init__ method is the default on or not
235		:rtype: boolean
236		"""
237	1	if '__objclass__' in dir(subject.__init__) and subject.__init__.__objclass__ == object:
238		return True
239	1	return False
240
241	1	def _get_argument(self, service, method_parameter):
242		"""
243		Retrieve the argument value for the given service
244
245		:param service: The service we need an argument for
246		:param method_parameter: The parameter we need the value for
247		:type service: Service
248		:type method_parameter: Parameter
249		:return: The argument value
250		:rtype: mixed
251		"""
252	1	for resolver in self._resolvers:
253	1	resolved = resolver.resolve(method_parameter, service)
254	1	if resolved:
255		return resolved
256
257		# If the parameter is args or *kwargs or has a default value
258		# then we don't raise any exception
259	1	if (method_parameter.kind in [Parameter.VAR_POSITIONAL, Parameter.VAR_KEYWORD]
260		or method_parameter.default is not Parameter.empty):
261	1	return None
262
263		self._logger.error("A required argument is not set: %s", method_parameter.name)
264		raise Exception("A required argument is not set (%s)" % method_parameter.name)
265

Darkheir / pyjection

GitHub Access Token became invalid

Branch — master (62458d)

DependencyInjector._generate_arguments_dict() B

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like