maxtepkeev / python-redmine / 317

pending completion

Build # 317

Build Type

push

travis-ci-com

Committed by

maxtepkeev

Commit Message

version bump

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

1366 of 1366 relevant lines covered (100.0%)

8.0 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

100.0

/redminelib/managers/base.py

"""
Defines base Redmine resource manager class and its infrastructure.
"""

from .. import resultsets, exceptions


class ResourceManager:
    """
    Manages given Redmine resource class with the help of redmine object.
    """
    def __init__(self, redmine, resource_class):
        """
        :param redmine.Redmine redmine: (required). Redmine object.
        :param resources.BaseResource resource_class: (required). Resource class.
        """
        self.url = ''
        self.params = {}
        self.container = None
        self.redmine = redmine
        self.resource_class = resource_class

    def to_resource(self, resource):
        """
        Converts resource data to Resource object.

        :param dict resource: (required). Resource data.
        """
        return self.resource_class(self, resource)

    def to_resource_set(self, resources):
        """
        Converts an iterable with resources data to ResourceSet object.

        :param resources: (required). Resource data.
        :type resources: list or tuple
        """
        return resultsets.ResourceSet(self, resources)

    def new(self):
        """
        Returns new empty Resource object.
        """
        return self.to_resource({})

    def new_manager(self, resource_name, **params):
        """
        Returns new ResourceManager object.

        :param string resource_name: (required). Resource name.
        :param dict params: (optional). Parameters used for resources retrieval.
        """
        manager = getattr(self.redmine, resource_name)
        manager.params = params
        return manager

    def _construct_get_url(self, path):
        """
        Constructs URL for get method.

        :param string path: absolute URL path.
        """
        return self.redmine.url + path

    def _prepare_get_request(self, request):
        """
        Makes the necessary preparations for get request data.

        :param dict request: Request data.
        """
        return self.resource_class.bulk_decode(request, self)

    def get(self, resource_id, **params):
        """
        Returns a Resource object from Redmine by resource id.

        :param resource_id: (required). Resource id.
        :type resource_id: int or string
        :param dict params: (optional). Parameters used for resource retrieval.
        """
        if self.resource_class.query_one is None or self.resource_class.container_one is None:
            operation = self.all if self.resource_class.query_all else self.filter
            resource = operation(**params).get(resource_id, None)

            if resource is None:
                raise exceptions.ResourceNotFoundError

            return resource

        try:
            self.url = self._construct_get_url(self.resource_class.query_one.format(resource_id, **params))
        except KeyError as e:
            raise exceptions.ValidationError(f'{e} argument is required')

        self.params = self._prepare_get_request(params)
        self.container = self.resource_class.container_one

        try:
            response = self.redmine.engine.request('get', self.url, params=self.params)
        except exceptions.ResourceNotFoundError as e:
            if self.resource_class.requirements:
                raise exceptions.ResourceRequirementsError(self.resource_class.requirements)
            raise e

        return self._process_get_response(self.params, response)

    def _process_get_response(self, request, response):
        """
        Processes get response and constructs resource object.

        :param dict request: Original request data.
        :param any response: Response received from Redmine for this request data.
        """
        return self.to_resource(response[self.container])

    def all(self, **params):
        """
        Returns a ResourceSet object with all Resource objects.

        :param dict params: (optional). Parameters used for resources retrieval.
        """
        if self.resource_class.query_all is None or self.resource_class.container_all is None:
            raise exceptions.ResourceBadMethodError

        self.url = self.redmine.url + self.resource_class.query_all
        self.params = self.resource_class.bulk_decode(params, self)
        self.container = self.resource_class.container_all
        return resultsets.ResourceSet(self)

    def filter(self, **filters):
        """
        Returns a ResourceSet object with Resource objects filtered by a dict of filters.

        :param dict filters: (optional). Filters used for resources retrieval.
        """
        if self.resource_class.query_filter is None or self.resource_class.container_filter is None:
            raise exceptions.ResourceBadMethodError

        if not filters:
            raise exceptions.ResourceNoFiltersProvidedError

        try:
            self.url = self.redmine.url + self.resource_class.query_filter.format(**filters)
            self.container = self.resource_class.container_filter.format(**filters)
        except KeyError:
            raise exceptions.ResourceFilterError

        self.params = self.resource_class.bulk_decode(filters, self)
        return resultsets.ResourceSet(self)

    def _construct_create_url(self, path):
        """
        Constructs URL for create method.

        :param string path: absolute URL path.
        """
        return self.redmine.url + path

    def _prepare_create_request(self, request):
        """
        Makes the necessary preparations for create request data.

        :param dict request: Request data.
        """
        return {self.container: self.resource_class.bulk_decode(request, self)}

    def create(self, **fields):
        """
        Creates a new resource in Redmine and returns created Resource object on success.

        :param dict fields: (optional). Fields used for resource creation.
        """
        if self.resource_class.query_create is None or self.resource_class.container_create is None:
            raise exceptions.ResourceBadMethodError

        if not fields:
            raise exceptions.ResourceNoFieldsProvidedError

        try:
            url = self._construct_create_url(self.resource_class.query_create.format(**fields))
        except KeyError as e:
            raise exceptions.ValidationError(f'{e} field is required')

        self.params = self.resource_class.query_create.formatter.used_kwargs
        self.container = self.resource_class.container_create
        request = self._prepare_create_request(self.resource_class.query_create.formatter.unused_kwargs)
        response = self.redmine.engine.request(self.resource_class.http_method_create, url, data=request)

        if response is None:
            return None

        resource = self._process_create_response(request, response)

        if self.resource_class.query_one is not None:
            self.url = self.redmine.url + self.resource_class.query_one.format(resource.internal_id, **fields)

        return resource

    def _process_create_response(self, request, response):
        """
        Processes create response and constructs resource object.

        :param dict request: Original request data.
        :param any response: Response received from Redmine for this request data.
        """
        return self.to_resource(response[self.container])

    def _construct_update_url(self, path):
        """
        Constructs URL for update method.

        :param string path: absolute URL path.
        """
        return self.redmine.url + path

    def _prepare_update_request(self, request):
        """
        Makes the necessary preparations for update request data.

        :param dict request: Request data.
        """
        return {self.resource_class.container_update: self.resource_class.bulk_decode(request, self)}

    def update(self, resource_id, **fields):
        """
        Updates a Resource object by resource id.

        :param resource_id: (required). Resource id.
        :type resource_id: int or string
        :param dict fields: (optional). Fields that will be updated for the resource.
        """
        if self.resource_class.query_update is None or self.resource_class.container_update is None:
            raise exceptions.ResourceBadMethodError

        if not fields:
            raise exceptions.ResourceNoFieldsProvidedError

        try:
            query_update = self.resource_class.query_update.format(resource_id, **fields)
        except KeyError as e:
            param = e.args[0]

            if param in self.params:
                fields[param] = self.params[param]
                query_update = self.resource_class.query_update.format(resource_id, **fields)
            else:
                raise exceptions.ValidationError(f'{e} argument is required')

        url = self._construct_update_url(query_update)
        request = self._prepare_update_request(self.resource_class.query_update.formatter.unused_kwargs)
        response = self.redmine.engine.request(self.resource_class.http_method_update, url, data=request)

        if response is None:
            return None

        return self._process_update_response(request, response)

    def _process_update_response(self, request, response):
        """
        Processes update response.

        :param dict request: Original request data.
        :param any response: Response received from Redmine for this request data.
        """
        return response

    def _construct_delete_url(self, path):
        """
        Constructs URL for delete method.

        :param string path: absolute URL path.
        """
        return self.redmine.url + path

    def _prepare_delete_request(self, request):
        """
        Makes the necessary preparations for delete request data.

        :param dict request: Request data.
        """
        return self.resource_class.bulk_decode(request, self)

    def delete(self, resource_id, **params):
        """
        Deletes a Resource object by resource id.

        :param resource_id: (required). Resource id.
        :type resource_id: int or string
        :param dict params: (optional). Parameters used for resource deletion.
        """
        if self.resource_class.query_delete is None:
            raise exceptions.ResourceBadMethodError

        try:
            url = self._construct_delete_url(self.resource_class.query_delete.format(resource_id, **params))
        except KeyError as e:
            raise exceptions.ValidationError(f'{e} argument is required')

        request = self._prepare_delete_request(params)
        response = self.redmine.engine.request(self.resource_class.http_method_delete, url, params=request)

        if response is None:
            return None

        return self._process_delete_response(request, response)

    def _process_delete_response(self, request, response):
        """
        Processes delete response.

        :param dict request: Original request data.
        :param any response: Response received from Redmine for this request data.
        """
        return response

    def search(self, query, **options):
        """
        Searches for Resources using a query.

        :param string query: (required). What to search.
        :param dict options: (optional). Dictionary of search options.
        """
        if self.resource_class.search_hints is None:
            raise exceptions.ResourceBadMethodError

        container = self.resource_class.container_all or self.resource_class.container_filter
        results = self.redmine.search(query, **dict(resources=[container], **options))
        return results.get(container) if results is not None else results

    def __repr__(self):
        """
        Official representation of a ResourceManager object.
        """
        return f'<redminelib.managers.{self.__class__.__name__} object for {self.resource_class.__name__} resource>'

1	"""
2	Defines base Redmine resource manager class and its infrastructure.
3	"""
4
5	from .. import resultsets, exceptions	8✔
6
7
8	class ResourceManager:	8✔
9	"""
10	Manages given Redmine resource class with the help of redmine object.
11	"""
12	def __init__(self, redmine, resource_class):	8✔
13	"""
14	:param redmine.Redmine redmine: (required). Redmine object.
15	:param resources.BaseResource resource_class: (required). Resource class.
16	"""
17	self.url = ''	8✔
18	self.params = {}	8✔
19	self.container = None	8✔
20	self.redmine = redmine	8✔
21	self.resource_class = resource_class	8✔
22
23	def to_resource(self, resource):	8✔
24	"""
25	Converts resource data to Resource object.
26
27	:param dict resource: (required). Resource data.
28	"""
29	return self.resource_class(self, resource)	8✔
30
31	def to_resource_set(self, resources):	8✔
32	"""
33	Converts an iterable with resources data to ResourceSet object.
34
35	:param resources: (required). Resource data.
36	:type resources: list or tuple
37	"""
38	return resultsets.ResourceSet(self, resources)	8✔
39
40	def new(self):	8✔
41	"""
42	Returns new empty Resource object.
43	"""
44	return self.to_resource({})	8✔
45
46	def new_manager(self, resource_name, **params):	8✔
47	"""
48	Returns new ResourceManager object.
49
50	:param string resource_name: (required). Resource name.
51	:param dict params: (optional). Parameters used for resources retrieval.
52	"""
53	manager = getattr(self.redmine, resource_name)	8✔
54	manager.params = params	8✔
55	return manager	8✔
56
57	def _construct_get_url(self, path):	8✔
58	"""
59	Constructs URL for get method.
60
61	:param string path: absolute URL path.
62	"""
63	return self.redmine.url + path	8✔
64
65	def _prepare_get_request(self, request):	8✔
66	"""
67	Makes the necessary preparations for get request data.
68
69	:param dict request: Request data.
70	"""
71	return self.resource_class.bulk_decode(request, self)	8✔
72
73	def get(self, resource_id, **params):	8✔
74	"""
75	Returns a Resource object from Redmine by resource id.
76
77	:param resource_id: (required). Resource id.
78	:type resource_id: int or string
79	:param dict params: (optional). Parameters used for resource retrieval.
80	"""
81	if self.resource_class.query_one is None or self.resource_class.container_one is None:	8✔
82	operation = self.all if self.resource_class.query_all else self.filter	8✔
83	resource = operation(**params).get(resource_id, None)	8✔
84
85	if resource is None:	8✔
86	raise exceptions.ResourceNotFoundError	8✔
87
88	return resource	8✔
89
90	try:	8✔
91	self.url = self._construct_get_url(self.resource_class.query_one.format(resource_id, **params))	8✔
92	except KeyError as e:	8✔
93	raise exceptions.ValidationError(f'{e} argument is required')	8✔
94
95	self.params = self._prepare_get_request(params)	8✔
96	self.container = self.resource_class.container_one	8✔
97
98	try:	8✔
99	response = self.redmine.engine.request('get', self.url, params=self.params)	8✔
100	except exceptions.ResourceNotFoundError as e:	8✔
101	if self.resource_class.requirements:	8✔
102	raise exceptions.ResourceRequirementsError(self.resource_class.requirements)	8✔
103	raise e	8✔
104
105	return self._process_get_response(self.params, response)	8✔
106
107	def _process_get_response(self, request, response):	8✔
108	"""
109	Processes get response and constructs resource object.
110
111	:param dict request: Original request data.
112	:param any response: Response received from Redmine for this request data.
113	"""
114	return self.to_resource(response[self.container])	8✔
115
116	def all(self, **params):	8✔
117	"""
118	Returns a ResourceSet object with all Resource objects.
119
120	:param dict params: (optional). Parameters used for resources retrieval.
121	"""
122	if self.resource_class.query_all is None or self.resource_class.container_all is None:	8✔
123	raise exceptions.ResourceBadMethodError	8✔
124
125	self.url = self.redmine.url + self.resource_class.query_all	8✔
126	self.params = self.resource_class.bulk_decode(params, self)	8✔
127	self.container = self.resource_class.container_all	8✔
128	return resultsets.ResourceSet(self)	8✔
129
130	def filter(self, **filters):	8✔
131	"""
132	Returns a ResourceSet object with Resource objects filtered by a dict of filters.
133
134	:param dict filters: (optional). Filters used for resources retrieval.
135	"""
136	if self.resource_class.query_filter is None or self.resource_class.container_filter is None:	8✔
137	raise exceptions.ResourceBadMethodError	8✔
138
139	if not filters:	8✔
140	raise exceptions.ResourceNoFiltersProvidedError	8✔
141
142	try:	8✔
143	self.url = self.redmine.url + self.resource_class.query_filter.format(**filters)	8✔
144	self.container = self.resource_class.container_filter.format(**filters)	8✔
145	except KeyError:	8✔
146	raise exceptions.ResourceFilterError	8✔
147
148	self.params = self.resource_class.bulk_decode(filters, self)	8✔
149	return resultsets.ResourceSet(self)	8✔
150
151	def _construct_create_url(self, path):	8✔
152	"""
153	Constructs URL for create method.
154
155	:param string path: absolute URL path.
156	"""
157	return self.redmine.url + path	8✔
158
159	def _prepare_create_request(self, request):	8✔
160	"""
161	Makes the necessary preparations for create request data.
162
163	:param dict request: Request data.
164	"""
165	return {self.container: self.resource_class.bulk_decode(request, self)}	8✔
166
167	def create(self, **fields):	8✔
168	"""
169	Creates a new resource in Redmine and returns created Resource object on success.
170
171	:param dict fields: (optional). Fields used for resource creation.
172	"""
173	if self.resource_class.query_create is None or self.resource_class.container_create is None:	8✔
174	raise exceptions.ResourceBadMethodError	8✔
175
176	if not fields:	8✔
177	raise exceptions.ResourceNoFieldsProvidedError	8✔
178
179	try:	8✔
180	url = self._construct_create_url(self.resource_class.query_create.format(**fields))	8✔
181	except KeyError as e:	8✔
182	raise exceptions.ValidationError(f'{e} field is required')	8✔
183
184	self.params = self.resource_class.query_create.formatter.used_kwargs	8✔
185	self.container = self.resource_class.container_create	8✔
186	request = self._prepare_create_request(self.resource_class.query_create.formatter.unused_kwargs)	8✔
187	response = self.redmine.engine.request(self.resource_class.http_method_create, url, data=request)	8✔
188
189	if response is None:	8✔
190	return None	8✔
191
192	resource = self._process_create_response(request, response)	8✔
193
194	if self.resource_class.query_one is not None:	8✔
195	self.url = self.redmine.url + self.resource_class.query_one.format(resource.internal_id, **fields)	8✔
196
197	return resource	8✔
198
199	def _process_create_response(self, request, response):	8✔
200	"""
201	Processes create response and constructs resource object.
202
203	:param dict request: Original request data.
204	:param any response: Response received from Redmine for this request data.
205	"""
206	return self.to_resource(response[self.container])	8✔
207
208	def _construct_update_url(self, path):	8✔
209	"""
210	Constructs URL for update method.
211
212	:param string path: absolute URL path.
213	"""
214	return self.redmine.url + path	8✔
215
216	def _prepare_update_request(self, request):	8✔
217	"""
218	Makes the necessary preparations for update request data.
219
220	:param dict request: Request data.
221	"""
222	return {self.resource_class.container_update: self.resource_class.bulk_decode(request, self)}	8✔
223
224	def update(self, resource_id, **fields):	8✔
225	"""
226	Updates a Resource object by resource id.
227
228	:param resource_id: (required). Resource id.
229	:type resource_id: int or string
230	:param dict fields: (optional). Fields that will be updated for the resource.
231	"""
232	if self.resource_class.query_update is None or self.resource_class.container_update is None:	8✔
233	raise exceptions.ResourceBadMethodError	8✔
234
235	if not fields:	8✔
236	raise exceptions.ResourceNoFieldsProvidedError	8✔
237
238	try:	8✔
239	query_update = self.resource_class.query_update.format(resource_id, **fields)	8✔
240	except KeyError as e:	8✔
241	param = e.args[0]	8✔
242
243	if param in self.params:	8✔
244	fields[param] = self.params[param]	8✔
245	query_update = self.resource_class.query_update.format(resource_id, **fields)	8✔
246	else:
247	raise exceptions.ValidationError(f'{e} argument is required')	8✔
248
249	url = self._construct_update_url(query_update)	8✔
250	request = self._prepare_update_request(self.resource_class.query_update.formatter.unused_kwargs)	8✔
251	response = self.redmine.engine.request(self.resource_class.http_method_update, url, data=request)	8✔
252
253	if response is None:	8✔
254	return None	8✔
255
256	return self._process_update_response(request, response)	8✔
257
258	def _process_update_response(self, request, response):	8✔
259	"""
260	Processes update response.
261
262	:param dict request: Original request data.
263	:param any response: Response received from Redmine for this request data.
264	"""
265	return response	8✔
266
267	def _construct_delete_url(self, path):	8✔
268	"""
269	Constructs URL for delete method.
270
271	:param string path: absolute URL path.
272	"""
273	return self.redmine.url + path	8✔
274
275	def _prepare_delete_request(self, request):	8✔
276	"""
277	Makes the necessary preparations for delete request data.
278
279	:param dict request: Request data.
280	"""
281	return self.resource_class.bulk_decode(request, self)	8✔
282
283	def delete(self, resource_id, **params):	8✔
284	"""
285	Deletes a Resource object by resource id.
286
287	:param resource_id: (required). Resource id.
288	:type resource_id: int or string
289	:param dict params: (optional). Parameters used for resource deletion.
290	"""
291	if self.resource_class.query_delete is None:	8✔
292	raise exceptions.ResourceBadMethodError	8✔
293
294	try:	8✔
295	url = self._construct_delete_url(self.resource_class.query_delete.format(resource_id, **params))	8✔
296	except KeyError as e:	8✔
297	raise exceptions.ValidationError(f'{e} argument is required')	8✔
298
299	request = self._prepare_delete_request(params)	8✔
300	response = self.redmine.engine.request(self.resource_class.http_method_delete, url, params=request)	8✔
301
302	if response is None:	8✔
303	return None	8✔
304
305	return self._process_delete_response(request, response)	8✔
306
307	def _process_delete_response(self, request, response):	8✔
308	"""
309	Processes delete response.
310
311	:param dict request: Original request data.
312	:param any response: Response received from Redmine for this request data.
313	"""
314	return response	8✔
315
316	def search(self, query, **options):	8✔
317	"""
318	Searches for Resources using a query.
319
320	:param string query: (required). What to search.
321	:param dict options: (optional). Dictionary of search options.
322	"""
323	if self.resource_class.search_hints is None:	8✔
324	raise exceptions.ResourceBadMethodError	8✔
325
326	container = self.resource_class.container_all or self.resource_class.container_filter	8✔
327	results = self.redmine.search(query, dict(resources=[container], options))	8✔
328	return results.get(container) if results is not None else results	8✔
329
330	def __repr__(self):	8✔
331	"""
332	Official representation of a ResourceManager object.
333	"""
334	return f'<redminelib.managers.{self.__class__.__name__} object for {self.resource_class.__name__} resource>'	8✔

maxtepkeev / python-redmine / 317

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous