DependencyInjector._get_instance() - Code Metrics - Inspection of "Adds concept of resolvers" - Darkheir/pyjection - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Branch — master (0bcf65)

by Raphaël

created 2017-12-26 14:42 UTC

DependencyInjector._get_instance() A

↳ Parent: DependencyInjector

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	5
CRAP Score	2

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
c	1
b	0
f	0
dl	0
loc	12
ccs	5
cts	5
cp	1
rs	9.4285
cc	2
crap	2

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)

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

        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)

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

        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 _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 then we don't raise any exception
        if method_parameter.kind == Parameter.VAR_POSITIONAL or \
                method_parameter.kind == Parameter.VAR_KEYWORD:
            return None
        # If the parameter has a default value then we don't raise any exception
        if 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	if self.has_service(identifier) is False:
105	1	self._logger.error("No service has been declared with ID %s", identifier)
106	1	raise Exception("No service has been declared with this ID")
107
108	1	service = self._services[identifier]
109	1	instance = self._get_singleton(identifier, service)
110	1	if instance is not None:
111	1	self._logger.debug("Return singleton with ID %s", identifier)
112	1	return instance
113
114	1	instance = self._get_instance(service)
115	1	self._set_singleton(identifier, instance, service)
116	1	self._logger.debug("Return instance with ID %s", identifier)
117	1	return instance
118
119	1	def get_uninstantiated(self, identifier):
120		if isinstance(identifier, str) is False:
121		identifier = get_service_subject_identifier(identifier)
122
123		if self.has_service(identifier) is False:
124		self._logger.error("No service has been declared with ID %s", identifier)
125		raise Exception("No service has been declared with this ID")
126
127		service = self._services[identifier]
128		return service.subject
129
130	1	def has_service(self, identifier):
131		"""
132		Check if the service matching the given identifier
133		has already been declared
134
135		:param identifier: Name of the service or the class
136		:type identifier: mixed
137		:return: Whether or not the service exists
138		:rtype: boolean
139		"""
140	1	if isinstance(identifier, str) is False:
141	1	identifier = get_service_subject_identifier(identifier)
142
143	1	if identifier in self._services:
144	1	return True
145	1	return False
146
147	1	def _get_singleton(self, identifier, service):
148		"""
149		Return the singleton if it has been setted and
150		the service represents a singleton
151
152		:param identifier: the singleton identifier
153		:param service: The service we need the singleton for
154		:type identifier: string
155		:type service: Service
156
157		:return: The singleton instance or None
158		:rtype: mixed
159		"""
160	1	if service.is_singleton is True and identifier in self._singletons:
161	1	return self._singletons[identifier]
162	1	return None
163
164	1	def _set_singleton(self, identifier, instance, service):
165		"""
166		Set the instance as a singleton in the dict
167		if the service represents a singleton
168
169		:param identifier: the singleton identifier
170		:param service: The service we want to set a singleton for
171		:param instance: The singleton instance
172		:type identifier: string
173		:type service: Service
174		:type instance: mixed
175		"""
176	1	if service.is_singleton is True:
177	1	self._singletons[identifier] = instance
178
179	1	def _get_instance(self, service):
180		"""
181		Return the instantiated object for the given service
182
183		:param service: The service we need an instance for
184		:type service: Service
185		:return: The instantiated object
186		"""
187	1	if service.type == 'instance':
188	1	return service.subject
189	1	arguments = self._generate_arguments_dict(service)
190	1	return service.subject(**arguments)
191
192	1	def _generate_arguments_dict(self, service):
193		"""
194		Generate a dict containing all the parameters values
195		required to Instantiate the service.
196
197		An exception is raised if a mandatory argument cannot be
198		retrieved.
199
200		:param service: The service that needs to be instantiated
201		:type service: Service
202		:return: The parameters values to use to instantiate the service
203		:rtype: dict
204		"""
205	1	arguments = dict()
206
207		# We can't use signature on class object __init__
208	1	if self._is_object_init(service.subject):
209		return arguments
210
211	1	sig = signature(service.subject.__init__)
212	1	method_parameters = OrderedDict(sig.parameters)
213
214		# Pop the first param since it's the self class instance
215	1	method_parameters.popitem(False)
216
217	1	for method_parameter in method_parameters.values():
218	1	argument = self._get_argument(service, method_parameter)
219	1	if argument is not None:
220		arguments[method_parameter.name] = argument
221
222	1	return arguments
223
224	1	@staticmethod
225		def _is_object_init(subject):
226		"""
227		Check if the __init__ method for the object comes from
228		the default object class or has been overridden
229
230		:param subject: The subject we want to check the __init__ for
231		:type subject: mixed
232
233		:return: Whether the __init__ method is the default on or not
234		:rtype: boolean
235		"""
236	1	if '__objclass__' in dir(subject.__init__) and subject.__init__.__objclass__ == object:
237		return True
238	1	return False
239
240	1	def _get_argument(self, service, method_parameter):
241		"""
242		Retrieve the argument value for the given service
243
244		:param service: The service we need an argument for
245		:param method_parameter: The parameter we need the value for
246		:type service: Service
247		:type method_parameter: Parameter
248		:return: The argument value
249		:rtype: mixed
250		"""
251	1	for resolver in self._resolvers:
252	1	resolved = resolver.resolve(method_parameter, service)
253	1	if resolved:
254		return resolved
255
256		# If the parameter is args or *kwargs then we don't raise any exception
257	1	if method_parameter.kind == Parameter.VAR_POSITIONAL or \
258		method_parameter.kind == Parameter.VAR_KEYWORD:
259	1	return None
260		# If the parameter has a default value then we don't raise any exception
261	1	if method_parameter.default is not Parameter.empty:
262	1	return None
263
264		self._logger.error("A required argument is not set: %s", method_parameter.name)
265		raise Exception("A required argument is not set (%s)" % method_parameter.name)
266

Darkheir / pyjection

GitHub Access Token became invalid

Branch — master (0bcf65)

DependencyInjector._get_instance() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like