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/__init__.py

"""
Provides public API.
"""

import os
import io
import inspect
import warnings
import datetime
import contextlib

from . import managers, exceptions, engines, utilities, resources
from .version import __version__


class Redmine:
    """
    Entry point for all requests.
    """
    def __init__(self, url, **kwargs):
        """
        :param string url: (required). Redmine location.
        :param string key: (optional). API key used for authentication.
        :param string version: (optional). Redmine version.
        :param string username: (optional). Username used for authentication.
        :param string password: (optional). Password used for authentication.
        :param dict requests: (optional). Connection options.
        :param string impersonate: (optional). Username to impersonate.
        :param string date_format: (optional). Formatting directives for date format.
        :param string datetime_format: (optional). Formatting directives for datetime format.
        :param raise_attr_exception: (optional). Control over resource attribute access exception raising.
        :type raise_attr_exception: bool or tuple
        :param timezone: (optional). Whether to convert a naive datetime to a specific timezone aware one.
        :type timezone: str or cls
        :param cls engine: (optional). Engine that will be used to make requests to Redmine.
        """
        self.url = url.rstrip('/')
        self.ver = kwargs.pop('version', None)

        if self.ver is not None:
            self.ver = utilities.versiontuple(self.ver)

        self.timezone = kwargs.pop('timezone', None)

        if self.timezone is not None and not isinstance(self.timezone, datetime.tzinfo):
            try:
                self.timezone = datetime.datetime.strptime(self.timezone, '%z').tzinfo
            except (TypeError, ValueError):
                raise exceptions.TimezoneError

        self.date_format = kwargs.pop('date_format', '%Y-%m-%d')
        self.datetime_format = kwargs.pop('datetime_format', '%Y-%m-%dT%H:%M:%SZ')
        self.raise_attr_exception = kwargs.pop('raise_attr_exception', True)

        engine = kwargs.pop('engine', engines.DefaultEngine)

        if not inspect.isclass(engine) or not issubclass(engine, engines.BaseEngine):
            raise exceptions.EngineClassError

        self.engine = engine(**kwargs)

    def __getattr__(self, resource_name):
        """
        Returns a ResourceManager object for the requested resource.

        :param string resource_name: (required). Resource name.
        """
        if resource_name.startswith('_'):
            raise AttributeError

        resource_name = ''.join(word[0].upper() + word[1:] for word in str(resource_name).split('_'))

        try:
            resource_class = resources.registry[resource_name]['class']
        except KeyError:
            raise exceptions.ResourceError

        if self.ver is not None and self.ver < resource_class.redmine_version:
            raise exceptions.ResourceVersionMismatchError

        return resource_class.manager_class(self, resource_class)

    @contextlib.contextmanager
    def session(self, **options):
        """
        Initiates a temporary session with a copy of the current engine but with new options.

        :param dict options: (optional). Engine's options for a session.
        """
        engine = self.engine
        self.engine = engine.__class__(
            requests=utilities.merge_dicts(engine.requests, options.pop('requests', {})), **options)

        try:
            yield self
        except exceptions.BaseRedmineError as e:
            raise e
        finally:
            self.engine = engine

    def upload(self, f, filename=None):
        """
        Uploads file from file path / file stream to Redmine and returns an assigned token.

        :param f: (required). File path / stream that will be uploaded.
        :type f: string or file-like object
        :param filename: (optional). Filename for the file that will be uploaded.
        """
        if self.ver is not None and self.ver < (1, 4, 0):
            raise exceptions.VersionMismatchError('File uploading')

        url = f'{self.url}/uploads.json'
        headers = {'Content-Type': 'application/octet-stream'}
        params = {'filename': filename or ''}

        # There are myriads of file-like object implementations here and there and some of them don't have
        # a "read" method, which is wrong, but that's what we have, on the other hand it looks like all of
        # them implement a "close" method, that's why we check for it here. Also, we don't want to close the
        # stream ourselves as we have no idea of what the client is going to do with it afterwards, so we
        # leave the closing part to the client or to the garbage collector
        if hasattr(f, 'close'):
            try:
                c = f.read(0)
            except (AttributeError, TypeError):
                raise exceptions.FileObjectError

            # We need to send bytes over the socket, so in case a file-like object contains a unicode
            # object underneath, we need to convert it to bytes, otherwise we'll get an exception
            if isinstance(c, str):
                warnings.warn("File-like object contains unicode, hence an additional step is performed to convert "
                              "its content to bytes, please consider switching to bytes to eliminate this warning",
                              exceptions.PerformanceWarning)
                f = io.BytesIO(f.read().encode('utf-8'))

            stream = f
            close = False
        else:
            if not os.path.isfile(f) or os.path.getsize(f) == 0:
                raise exceptions.NoFileError

            stream = open(f, 'rb')
            close = True

        response = self.engine.request('post', url, params=params, data=stream, headers=headers)

        if close:
            stream.close()

        return response['upload']

    def download(self, url, savepath=None, filename=None, params=None):
        """
        Downloads file from Redmine and saves it to savepath or returns a response directly
        for maximum control over file processing.

        :param string url: (required). URL of the file that will be downloaded.
        :param string savepath: (optional). Path where to save the file.
        :param string filename: (optional). Name that will be used for the file.
        :param dict params: (optional). Params to send in the query string.
        """
        with self.session(requests={'stream': True}, return_raw_response=True):
            response = self.engine.request('get', url, params=params or {})

        # If a savepath wasn't provided we return a response directly
        # so a user can have maximum control over response data
        if savepath is None:
            return response

        from urllib.parse import urlsplit

        if filename is None:
            filename = urlsplit(url)[2].split('/')[-1]

            if not filename:
                raise exceptions.FileUrlError

        savepath = os.path.join(savepath, filename)

        with open(savepath, 'wb') as f:
            for chunk in response.iter_content(1024):
                f.write(chunk)

        return savepath

    def auth(self):
        """
        Shortcut for the case if we just want to check if user provided valid auth credentials.
        """
        return self.user.get('current')

    def search(self, query, **options):
        """
        Interface to Redmine Search API

        :param string query: (required). What to search.
        :param dict options: (optional). Dictionary of search options.
        """
        if self.ver is not None and self.ver < (3, 0, 0):
            raise exceptions.VersionMismatchError('Search functionality')

        container_map, manager_map, results = {}, {}, {'unknown': {}}

        for resource in options.pop('resources', []):
            options[resource] = True

        options['q'] = query

        for name, details in resources.registry.items():
            if details['class'].search_hints is not None:
                container = details['class'].container_all or details['class'].container_filter

                for hint in details['class'].search_hints:
                    container_map[hint] = container

                manager_map[container] = getattr(self, name)

        raw_resources, _ = self.engine.bulk_request('get', f'{self.url}/search.json', 'results', **options)

        for resource in raw_resources:
            if resource['type'] in container_map:
                container = container_map[resource['type']]

                if container not in results:
                    results[container] = []

                results[container].append(resource)
            else:
                if resource['type'] not in results['unknown']:
                    results['unknown'][resource['type']] = []

                results['unknown'][resource['type']].append(resource)

            del resource['type']  # all resources are already sorted by type, so we don't need it

        if not results['unknown']:
            del results['unknown']

        for container in results:
            if container in manager_map:
                results[container] = manager_map[container].to_resource_set(results[container])

        return results or None

1	"""
2	Provides public API.
3	"""
4
5	import os	8✔
6	import io	8✔
7	import inspect	8✔
8	import warnings	8✔
9	import datetime	8✔
10	import contextlib	8✔
11
12	from . import managers, exceptions, engines, utilities, resources	8✔
13	from .version import __version__	8✔
14
15
16	class Redmine:	8✔
17	"""
18	Entry point for all requests.
19	"""
20	def __init__(self, url, **kwargs):	8✔
21	"""
22	:param string url: (required). Redmine location.
23	:param string key: (optional). API key used for authentication.
24	:param string version: (optional). Redmine version.
25	:param string username: (optional). Username used for authentication.
26	:param string password: (optional). Password used for authentication.
27	:param dict requests: (optional). Connection options.
28	:param string impersonate: (optional). Username to impersonate.
29	:param string date_format: (optional). Formatting directives for date format.
30	:param string datetime_format: (optional). Formatting directives for datetime format.
31	:param raise_attr_exception: (optional). Control over resource attribute access exception raising.
32	:type raise_attr_exception: bool or tuple
33	:param timezone: (optional). Whether to convert a naive datetime to a specific timezone aware one.
34	:type timezone: str or cls
35	:param cls engine: (optional). Engine that will be used to make requests to Redmine.
36	"""
37	self.url = url.rstrip('/')	8✔
38	self.ver = kwargs.pop('version', None)	8✔
39
40	if self.ver is not None:	8✔
41	self.ver = utilities.versiontuple(self.ver)	8✔
42
43	self.timezone = kwargs.pop('timezone', None)	8✔
44
45	if self.timezone is not None and not isinstance(self.timezone, datetime.tzinfo):	8✔
46	try:	8✔
47	self.timezone = datetime.datetime.strptime(self.timezone, '%z').tzinfo	8✔
48	except (TypeError, ValueError):	8✔
49	raise exceptions.TimezoneError	8✔
50
51	self.date_format = kwargs.pop('date_format', '%Y-%m-%d')	8✔
52	self.datetime_format = kwargs.pop('datetime_format', '%Y-%m-%dT%H:%M:%SZ')	8✔
53	self.raise_attr_exception = kwargs.pop('raise_attr_exception', True)	8✔
54
55	engine = kwargs.pop('engine', engines.DefaultEngine)	8✔
56
57	if not inspect.isclass(engine) or not issubclass(engine, engines.BaseEngine):	8✔
58	raise exceptions.EngineClassError	8✔
59
60	self.engine = engine(**kwargs)	8✔
61
62	def __getattr__(self, resource_name):	8✔
63	"""
64	Returns a ResourceManager object for the requested resource.
65
66	:param string resource_name: (required). Resource name.
67	"""
68	if resource_name.startswith('_'):	8✔
69	raise AttributeError	8✔
70
71	resource_name = ''.join(word[0].upper() + word[1:] for word in str(resource_name).split('_'))	8✔
72
73	try:	8✔
74	resource_class = resources.registry[resource_name]['class']	8✔
75	except KeyError:	8✔
76	raise exceptions.ResourceError	8✔
77
78	if self.ver is not None and self.ver < resource_class.redmine_version:	8✔
79	raise exceptions.ResourceVersionMismatchError	8✔
80
81	return resource_class.manager_class(self, resource_class)	8✔
82
83	@contextlib.contextmanager	8✔
84	def session(self, **options):	5✔
85	"""
86	Initiates a temporary session with a copy of the current engine but with new options.
87
88	:param dict options: (optional). Engine's options for a session.
89	"""
90	engine = self.engine	8✔
91	self.engine = engine.__class__(	8✔
92	requests=utilities.merge_dicts(engine.requests, options.pop('requests', {})), **options)
93
94	try:	8✔
95	yield self	8✔
96	except exceptions.BaseRedmineError as e:	8✔
97	raise e	8✔
98	finally:
99	self.engine = engine	8✔
100
101	def upload(self, f, filename=None):	8✔
102	"""
103	Uploads file from file path / file stream to Redmine and returns an assigned token.
104
105	:param f: (required). File path / stream that will be uploaded.
106	:type f: string or file-like object
107	:param filename: (optional). Filename for the file that will be uploaded.
108	"""
109	if self.ver is not None and self.ver < (1, 4, 0):	8✔
110	raise exceptions.VersionMismatchError('File uploading')	8✔
111
112	url = f'{self.url}/uploads.json'	8✔
113	headers = {'Content-Type': 'application/octet-stream'}	8✔
114	params = {'filename': filename or ''}	8✔
115
116	# There are myriads of file-like object implementations here and there and some of them don't have
117	# a "read" method, which is wrong, but that's what we have, on the other hand it looks like all of
118	# them implement a "close" method, that's why we check for it here. Also, we don't want to close the
119	# stream ourselves as we have no idea of what the client is going to do with it afterwards, so we
120	# leave the closing part to the client or to the garbage collector
121	if hasattr(f, 'close'):	8✔
122	try:	8✔
123	c = f.read(0)	8✔
124	except (AttributeError, TypeError):	8✔
125	raise exceptions.FileObjectError	8✔
126
127	# We need to send bytes over the socket, so in case a file-like object contains a unicode
128	# object underneath, we need to convert it to bytes, otherwise we'll get an exception
129	if isinstance(c, str):	8✔
130	warnings.warn("File-like object contains unicode, hence an additional step is performed to convert "	8✔
131	"its content to bytes, please consider switching to bytes to eliminate this warning",
132	exceptions.PerformanceWarning)
133	f = io.BytesIO(f.read().encode('utf-8'))	8✔
134
135	stream = f	8✔
136	close = False	8✔
137	else:
138	if not os.path.isfile(f) or os.path.getsize(f) == 0:	8✔
139	raise exceptions.NoFileError	8✔
140
141	stream = open(f, 'rb')	8✔
142	close = True	8✔
143
144	response = self.engine.request('post', url, params=params, data=stream, headers=headers)	8✔
145
146	if close:	8✔
147	stream.close()	8✔
148
149	return response['upload']	8✔
150
151	def download(self, url, savepath=None, filename=None, params=None):	8✔
152	"""
153	Downloads file from Redmine and saves it to savepath or returns a response directly
154	for maximum control over file processing.
155
156	:param string url: (required). URL of the file that will be downloaded.
157	:param string savepath: (optional). Path where to save the file.
158	:param string filename: (optional). Name that will be used for the file.
159	:param dict params: (optional). Params to send in the query string.
160	"""
161	with self.session(requests={'stream': True}, return_raw_response=True):	8✔
162	response = self.engine.request('get', url, params=params or {})	8✔
163
164	# If a savepath wasn't provided we return a response directly
165	# so a user can have maximum control over response data
166	if savepath is None:	8✔
167	return response	8✔
168
169	from urllib.parse import urlsplit	8✔
170
171	if filename is None:	8✔
172	filename = urlsplit(url)[2].split('/')[-1]	8✔
173
174	if not filename:	8✔
175	raise exceptions.FileUrlError	8✔
176
177	savepath = os.path.join(savepath, filename)	8✔
178
179	with open(savepath, 'wb') as f:	8✔
180	for chunk in response.iter_content(1024):	8✔
181	f.write(chunk)	8✔
182
183	return savepath	8✔
184
185	def auth(self):	8✔
186	"""
187	Shortcut for the case if we just want to check if user provided valid auth credentials.
188	"""
189	return self.user.get('current')	8✔
190
191	def search(self, query, **options):	8✔
192	"""
193	Interface to Redmine Search API
194
195	:param string query: (required). What to search.
196	:param dict options: (optional). Dictionary of search options.
197	"""
198	if self.ver is not None and self.ver < (3, 0, 0):	8✔
199	raise exceptions.VersionMismatchError('Search functionality')	8✔
200
201	container_map, manager_map, results = {}, {}, {'unknown': {}}	8✔
202
203	for resource in options.pop('resources', []):	8✔
204	options[resource] = True	8✔
205
206	options['q'] = query	8✔
207
208	for name, details in resources.registry.items():	8✔
209	if details['class'].search_hints is not None:	8✔
210	container = details['class'].container_all or details['class'].container_filter	8✔
211
212	for hint in details['class'].search_hints:	8✔
213	container_map[hint] = container	8✔
214
215	manager_map[container] = getattr(self, name)	8✔
216
217	raw_resources, _ = self.engine.bulk_request('get', f'{self.url}/search.json', 'results', **options)	8✔
218
219	for resource in raw_resources:	8✔
220	if resource['type'] in container_map:	8✔
221	container = container_map[resource['type']]	8✔
222
223	if container not in results:	8✔
224	results[container] = []	8✔
225
226	results[container].append(resource)	8✔
227	else:
228	if resource['type'] not in results['unknown']:	8✔
229	results['unknown'][resource['type']] = []	8✔
230
231	results['unknown'][resource['type']].append(resource)	8✔
232
233	del resource['type'] # all resources are already sorted by type, so we don't need it	8✔
234
235	if not results['unknown']:	8✔
236	del results['unknown']	8✔
237
238	for container in results:	8✔
239	if container in manager_map:	8✔
240	results[container] = manager_map[container].to_resource_set(results[container])	8✔
241
242	return results or None	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