DependencyInjector._get_argument() - Code Metrics - Inspection of "Improves the test coverage" - Darkheir/pyjection - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 2f370c...296b06 )

by Raphaël

created 2018-01-04 11:24 UTC

DependencyInjector._get_argument() B

↳ Parent: DependencyInjector

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	10
CRAP Score	5

Importance

Changes	6
Bugs	0	Features	0

Metric	Value
c	6
b	0
f	0
dl	0
loc	25
ccs	10
cts	10
cp	1
rs	8.0894
cc	5
crap	5

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
        """
        identifier = self._get_string_identifier(identifier)

        self._validate_service_name(identifier)

        service = self._services[identifier]
        instance = self._get_singleton(identifier, service)
        if instance:
            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):
        identifier = self._get_string_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_string_identifier(self, identifier):
        if isinstance(identifier, str):
            return identifier
        return get_service_subject_identifier(identifier)

    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

        error = "A required argument is not set: {0}".format(method_parameter.name)
        self._logger.error(error)
        raise Exception(error)


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

Darkheir / pyjection

GitHub Access Token became invalid

Push — master ( 2f370c...296b06 )

DependencyInjector._get_argument() B

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like