pyUSIrest.client.Submission.get_samples() - Code Metrics - Inspection of ":bookmark: pyUSIrest release v0.3.0" - cnr-ibba/pyUSIrest - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#7)

by Paolo

created 2020-01-14 11:02 UTC

pyUSIrest.client.Submission.get_samples() B

↳ Parent: Project

Complexity

Conditions

Size

Total Lines	51
Code Lines	22

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	7
eloc	22
nop	4
dl	0
loc	51
rs	7.952
c	0
b	0
f	0

How to fix Long Method

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Created on Thu Dec 19 16:28:46 2019

@author: Paolo Cozzi <[email protected]>
"""

import requests
import logging

from dateutil.parser import parse as parse_date

from . import __version__
from .auth import Auth
from .exceptions import USIConnectionError, TokenExpiredError

logger = logging.getLogger(__name__)


# https://stackoverflow.com/a/25341965/4385116
def is_date(string, fuzzy=False):
    """
    Return whether the string can be interpreted as a date.

    :param string: str, string to check for date
    :param fuzzy: bool, ignore unknown tokens in string if True
    """
    try:
        parse_date(string, fuzzy=fuzzy)
        return True

    except ValueError:
        return False


class Client():
    """A class to deal with EBI submission API. It perform request
    modelling user token in request headers. You need to call this class after
    instantiating an :py:class:`Auth <pyUSIrest.auth.Auth>` object::

        import getpass
        from pyUSIrest.auth import Auth
        from pyUSIrest.client import Client
        auth = Auth(user=<you_aap_user>, password=getpass.getpass())
        client = Client(auth)
        response = client.get("https://submission-test.ebi.ac.uk/api/")

    Attributes:
        headers (dict): default headers for requests
        last_response (requests.Response): last response object read by this
            class
        last_satus_code (int): last status code read by this class
        session (request.Session): a session object
        auth (Auth): a pyUSIrest Auth object
    """

    headers = {
        'Accept': 'application/hal+json',
        'User-Agent': 'pyUSIrest %s' % (__version__)
    }

    def __init__(self, auth):
        """Instantiate the class

        Args:
            auth (Auth): a valid :py:class:`Auth <pyUSIrest.auth.Auth>` object

        """

        # my attributes
        self._auth = None
        self.last_response = None
        self.last_status_code = None
        self.session = requests.Session()

        # setting auth object
        self.auth = auth

    @property
    def auth(self):
        """Get/Set :py:class:`Auth <pyUSIrest.auth.Auth>` object"""

        return self._auth

    @auth.setter
    def auth(self, auth):
        logger.debug("Auth type is %s" % (type(auth)))

        # assign Auth object or create a new one
        if isinstance(auth, Auth):
            logger.debug("Assigning an Auth object")
            self._auth = auth

        else:
            logger.debug("Creating an Auth object")
            self._auth = Auth(token=auth)

        logger.debug("Updating headers with token")
        self.headers['Authorization'] = "Bearer {token}".format(
            token=self._auth.token)

    def check_headers(self, headers=None):
        """Checking headers and token

        Args:
            headers (dict): custom header for request

        Returns:
            headers (dict): an update headers tocken"""

        if self.auth.is_expired():
            raise TokenExpiredError("Your token is expired")

        if not headers:
            logger.debug("Using default headers")
            new_headers = self.headers

        else:
            new_headers = self.headers.copy()
            new_headers.update(headers)

        return new_headers

    def check_status(self, response, expected_status=200):
        """Check response status. See `HTTP status codes <https://submission.
        ebi.ac.uk/api/docs/ref_overview.html#_http_status_codes>`_

        Args:
            response (requests.Reponse): the reponse returned by requests
            method
        """

        # check with status code. deal with 50X statuses (internal error)
        if int(response.status_code / 100) == 5:
            raise USIConnectionError(
                "Problems with API endpoints: %s" % response.text)

        # TODO: evaluate a list of expected status?
        if response.status_code != expected_status:
            raise USIConnectionError(
                "%s:%s" % (response.status_code, response.text))

    def get(self, url, headers={}, params={}):

        """Generic GET method

        Args:
            url (str): url to request
            headers (dict): custom headers for get request
            params (dict): custom params for get request

        Returns:
            requests.Response: a response object
        """

        logger.debug("Getting %s" % (url))
        headers = self.check_headers(headers)
        response = self.session.get(url, headers=headers, params=params)

        # track last response
        self.last_response = response
        self.last_status_code = response.status_code

        # check response status code
        self.check_status(response)

        return response

    def post(self, url, payload={}, headers={}, params={}):
        """Generic POST method

        Args:
            url (str): url to request
            payload (dict): data to send
            headers (dict): custom header for request
            params (dict): custom params for request

        Returns:
            requests.Response: a response object
        """

        logger.debug("Posting data to %s" % (url))
        headers = self.check_headers(headers)
        response = self.session.post(
            url, json=payload, headers=headers, params=params)

        # track last response
        self.last_response = response
        self.last_status_code = response.status_code

        # check response status code
        self.check_status(response, expected_status=201)

        return response

    def patch(self, url, payload={}, headers={}, params={}):
        """Generic PATCH method

        Args:
            url (str): url to request
            payload (dict): data to send
            headers (dict): custom header for request
            params (dict): custom params for request

        Returns:
            requests.Response: a response object
        """

        logger.debug("Patching data to %s" % (url))
        headers = self.check_headers(headers)
        response = self.session.patch(
            url, json=payload, headers=headers, params=params)

        # track last response
        self.last_response = response
        self.last_status_code = response.status_code

        # check response status code
        self.check_status(response)

        return response

    def delete(self, url, headers={}, params={}):
        """Generic DELETE method

        Args:
            url (str): url to request
            headers (dict): custom header for request
            params (dict): custom params for request

        Returns:
            requests.Response: a response object
        """

        logger.debug("Deleting %s" % (url))

        headers = self.check_headers(headers)
        response = self.session.delete(url, headers=headers, params=params)

        # track last response
        self.last_response = response
        self.last_status_code = response.status_code

        # check response status code
        self.check_status(response, expected_status=204)

        return response

    def put(self, url, payload={}, headers={}, params={}):

        """Generic PUT method

        Args:
            url (str): url to request
            payload (dict): data to send
            params (dict): custom params for request
            headers (dict): custom header for request

        Returns:
            requests.Response: a response object
        """

        logger.debug("Putting data to %s" % (url))
        headers = self.check_headers(headers)
        response = self.session.put(
            url, json=payload, headers=headers, params=params)

        # track last response
        self.last_response = response
        self.last_status_code = response.status_code

        # check response status code
        self.check_status(response)

        return response


class Document(Client):
    """Base class for pyUSIrest classes. It models common methods and
    attributes by calling :py:class:`Client` and reading json response from
    biosample API

    Attributes:
        _link (dict): ``_links`` data read from USI response
        _embeddedd (dict): ``_embedded`` data read from USI response
        page (dict): ``page`` data read from USI response
        name (str): name of this object
        data (dict): data from USI read with
            :py:meth:`response.json() <requests.Response.json>`

    """

    def __init__(self, auth=None, data=None):
        # if I get auth, setting appropriate method
        if auth:
            Client.__init__(self, auth)

        # my class attributes
        self._links = {}
        self._embedded = {}
        self.page = {}
        self.name = None
        self.data = {}

        # if I get data, read data into myself
        if data:
            self.read_data(data)

    def get(self, url, force_keys=True):
        """Override the Client.get method and read data into object::

            document = Document(auth)
            document.get(settings.ROOT_URL + "/api/")

        Args:
            url (str): url to request
            force_keys (bool): If True, define a new class attribute from data
                keys

        Returns:
            requests.Response: a response object
        """

        # call the base method
        response = super().get(url)

        # read data
        self.read_data(response.json(), force_keys)

        # act like client object
        return response

    def read_data(self, data, force_keys=False):
        """Read data from a dictionary object and set class attributes

        Args:
            data (dict): a data dictionary object read with
                :py:meth:`response.json() <requests.Response.json>`
            force_keys (bool): If True, define a new class attribute from data
                keys
        """

        # dealing with this type of documents
        for key in data.keys():
            if "date" in key.lower() and is_date(data[key]):
                self.__update_key(key, parse_date(data[key]), force_keys)

            else:
                self.__update_key(key, data[key], force_keys)

        self.data = data

    def __update_key(self, key, value, force_keys=False):
        """Helper function to update keys"""

        if hasattr(self, key):
            if getattr(self, key) and getattr(self, key) != '':
                # when I reload data, I do a substitution
                logger.debug("Found %s -> %s" % (key, getattr(self, key)))
                logger.debug("Updating %s -> %s" % (key, value))

            else:
                # don't have this attribute set
                logger.debug("Setting %s -> %s" % (key, value))

            setattr(self, key, value)

        else:
            if force_keys is True:
                logger.debug("Forcing %s -> %s" % (key, value))
                setattr(self, key, value)

            else:
                logger.warning("key %s not implemented" % (key))

    @classmethod
    def clean_url(cls, url):
        """Remove stuff like ``{?projection}`` from url

        Args:
            url (str): a string url

        Returns:
            str: the cleaned url
        """

        # remove {?projection} from self url. This is unreachable
        if '{?projection}' in url:
            logger.debug("removing {?projection} from url")
            url = url.replace("{?projection}", "")

        return url

    @classmethod
    def read_url(cls, auth, url):
        """Read a url and returns a :py:class:`Document` object

        Args:
            auth (Auth): an Auth object to pass to result
            url (str): url to request

        Returns:
            Document: a document object
        """

        # clean url
        url = cls.clean_url(url)

        # create a new document
        document = cls(auth=auth)

        # get url and load data
        document.get(url)

        return document

    def paginate(self):
        """Follow all the pages. Return an iterator of document objects

        Args:
            response (requests.Response): a response object

        Yield:
            Document: a new Document instance
        """

        # return myself
        yield self

        # track the current document
        document = self

        while 'next' in document._links:
            url = document._links['next']['href']
            document = Document.read_url(self.auth, url)

            # return the last document
            yield document

    def follow_tag(self, tag, force_keys=True):
        """Pick a url from data attribute relying on tag, perform a request
        and returns a document object. For instance::

            document.follow_tag('userSubmissions')

        will return a document instance by requesting with
        :py:meth:`Client.get` using
        ``document._links['userSubmissions']['href']`` as url

        Args:
            tag (str): a key from USI response dictionary
            force_keys (bool): set a new class attribute if not present

        Returns:
            Document: a document object
        """

        logger.debug("Following %s url" % (tag))

        url = self._links[tag]['href']

        # create a new document
        document = Document(auth=self.auth)

        # read data
        document.get(url, force_keys)

        return document

    def follow_self_url(self):
        """Follow *self* url and update class attributes. For instance::

            document.follow_self_url()

        will reload document instance by requesting with
        :py:meth:`Client.get` using
        ``document.data['_links']['self']['href']`` as url"""

        logger.debug("Following self url")

        # get a url to follow
        url = self._links['self']['href']

        # clean url
        url = self.clean_url(url)

        logger.debug("Updating self")

        # now follow self url and load data to self
        self.get(url)


1		#!/usr/bin/env python3
2		# -- coding: utf-8 --
3		"""
4		Created on Thu Dec 19 16:28:46 2019
5
6		@author: Paolo Cozzi <[email protected]>
7		"""
8
9		import requests
10		import logging
11
12		from dateutil.parser import parse as parse_date
13
14		from . import __version__
15		from .auth import Auth
16		from .exceptions import USIConnectionError, TokenExpiredError
17
18		logger = logging.getLogger(__name__)
19
20
21		# https://stackoverflow.com/a/25341965/4385116
22		def is_date(string, fuzzy=False):
23		"""
24		Return whether the string can be interpreted as a date.
25
26		:param string: str, string to check for date
27		:param fuzzy: bool, ignore unknown tokens in string if True
28		"""
29		try:
30		parse_date(string, fuzzy=fuzzy)
31		return True
32
33		except ValueError:
34		return False
35
36
37		class Client():
38		"""A class to deal with EBI submission API. It perform request
39		modelling user token in request headers. You need to call this class after
40		instantiating an :py:class:`Auth <pyUSIrest.auth.Auth>` object::
41
42		import getpass
43		from pyUSIrest.auth import Auth
44		from pyUSIrest.client import Client
45		auth = Auth(user=<you_aap_user>, password=getpass.getpass())
46		client = Client(auth)
47		response = client.get("https://submission-test.ebi.ac.uk/api/")
48
49		Attributes:
50		headers (dict): default headers for requests
51		last_response (requests.Response): last response object read by this
52		class
53		last_satus_code (int): last status code read by this class
54		session (request.Session): a session object
55		auth (Auth): a pyUSIrest Auth object
56		"""
57
58		headers = {
59		'Accept': 'application/hal+json',
60		'User-Agent': 'pyUSIrest %s' % (__version__)
61		}
62
63		def __init__(self, auth):
64		"""Instantiate the class
65
66		Args:
67		auth (Auth): a valid :py:class:`Auth <pyUSIrest.auth.Auth>` object
68
69		"""
70
71		# my attributes
72		self._auth = None
73		self.last_response = None
74		self.last_status_code = None
75		self.session = requests.Session()
76
77		# setting auth object
78		self.auth = auth
79
80		@property
81		def auth(self):
82		"""Get/Set :py:class:`Auth <pyUSIrest.auth.Auth>` object"""
83
84		return self._auth
85
86		@auth.setter
87		def auth(self, auth):
88		logger.debug("Auth type is %s" % (type(auth)))
89
90		# assign Auth object or create a new one
91		if isinstance(auth, Auth):
92		logger.debug("Assigning an Auth object")
93		self._auth = auth
94
95		else:
96		logger.debug("Creating an Auth object")
97		self._auth = Auth(token=auth)
98
99		logger.debug("Updating headers with token")
100		self.headers['Authorization'] = "Bearer {token}".format(
101		token=self._auth.token)
102
103		def check_headers(self, headers=None):
104		"""Checking headers and token
105
106		Args:
107		headers (dict): custom header for request
108
109		Returns:
110		headers (dict): an update headers tocken"""
111
112		if self.auth.is_expired():
113		raise TokenExpiredError("Your token is expired")
114
115		if not headers:
116		logger.debug("Using default headers")
117		new_headers = self.headers
118
119		else:
120		new_headers = self.headers.copy()
121		new_headers.update(headers)
122
123		return new_headers
124
125		def check_status(self, response, expected_status=200):
126		"""Check response status. See `HTTP status codes <https://submission.
127		ebi.ac.uk/api/docs/ref_overview.html#_http_status_codes>`_
128
129		Args:
130		response (requests.Reponse): the reponse returned by requests
131		method
132		"""
133
134		# check with status code. deal with 50X statuses (internal error)
135		if int(response.status_code / 100) == 5:
136		raise USIConnectionError(
137		"Problems with API endpoints: %s" % response.text)
138
139		# TODO: evaluate a list of expected status?
140		if response.status_code != expected_status:
141		raise USIConnectionError(
142		"%s:%s" % (response.status_code, response.text))
143
144	View Code Duplication	def get(self, url, headers={}, params={}):
		0 ignored issues – show Duplication introduced 2020-01-09 13:44 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
145		"""Generic GET method
146
147		Args:
148		url (str): url to request
149		headers (dict): custom headers for get request
150		params (dict): custom params for get request
151
152		Returns:
153		requests.Response: a response object
154		"""
155
156		logger.debug("Getting %s" % (url))
157		headers = self.check_headers(headers)
158		response = self.session.get(url, headers=headers, params=params)
159
160		# track last response
161		self.last_response = response
162		self.last_status_code = response.status_code
163
164		# check response status code
165		self.check_status(response)
166
167		return response
168
169		def post(self, url, payload={}, headers={}, params={}):
170		"""Generic POST method
171
172		Args:
173		url (str): url to request
174		payload (dict): data to send
175		headers (dict): custom header for request
176		params (dict): custom params for request
177
178		Returns:
179		requests.Response: a response object
180		"""
181
182		logger.debug("Posting data to %s" % (url))
183		headers = self.check_headers(headers)
184		response = self.session.post(
185		url, json=payload, headers=headers, params=params)
186
187		# track last response
188		self.last_response = response
189		self.last_status_code = response.status_code
190
191		# check response status code
192		self.check_status(response, expected_status=201)
193
194		return response
195
196		def patch(self, url, payload={}, headers={}, params={}):
197		"""Generic PATCH method
198
199		Args:
200		url (str): url to request
201		payload (dict): data to send
202		headers (dict): custom header for request
203		params (dict): custom params for request
204
205		Returns:
206		requests.Response: a response object
207		"""
208
209		logger.debug("Patching data to %s" % (url))
210		headers = self.check_headers(headers)
211		response = self.session.patch(
212		url, json=payload, headers=headers, params=params)
213
214		# track last response
215		self.last_response = response
216		self.last_status_code = response.status_code
217
218		# check response status code
219		self.check_status(response)
220
221		return response
222
223		def delete(self, url, headers={}, params={}):
224		"""Generic DELETE method
225
226		Args:
227		url (str): url to request
228		headers (dict): custom header for request
229		params (dict): custom params for request
230
231		Returns:
232		requests.Response: a response object
233		"""
234
235		logger.debug("Deleting %s" % (url))
236
237		headers = self.check_headers(headers)
238		response = self.session.delete(url, headers=headers, params=params)
239
240		# track last response
241		self.last_response = response
242		self.last_status_code = response.status_code
243
244		# check response status code
245		self.check_status(response, expected_status=204)
246
247		return response
248
249	View Code Duplication	def put(self, url, payload={}, headers={}, params={}):
		0 ignored issues – show Duplication introduced 2020-01-09 13:44 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
250		"""Generic PUT method
251
252		Args:
253		url (str): url to request
254		payload (dict): data to send
255		params (dict): custom params for request
256		headers (dict): custom header for request
257
258		Returns:
259		requests.Response: a response object
260		"""
261
262		logger.debug("Putting data to %s" % (url))
263		headers = self.check_headers(headers)
264		response = self.session.put(
265		url, json=payload, headers=headers, params=params)
266
267		# track last response
268		self.last_response = response
269		self.last_status_code = response.status_code
270
271		# check response status code
272		self.check_status(response)
273
274		return response
275
276
277		class Document(Client):
278		"""Base class for pyUSIrest classes. It models common methods and
279		attributes by calling :py:class:`Client` and reading json response from
280		biosample API
281
282		Attributes:
283		_link (dict): ``_links`` data read from USI response
284		_embeddedd (dict): ``_embedded`` data read from USI response
285		page (dict): ``page`` data read from USI response
286		name (str): name of this object
287		data (dict): data from USI read with
288		:py:meth:`response.json() <requests.Response.json>`
289
290		"""
291
292		def __init__(self, auth=None, data=None):
293		# if I get auth, setting appropriate method
294		if auth:
295		Client.__init__(self, auth)
296
297		# my class attributes
298		self._links = {}
299		self._embedded = {}
300		self.page = {}
301		self.name = None
302		self.data = {}
303
304		# if I get data, read data into myself
305		if data:
306		self.read_data(data)
307
308		def get(self, url, force_keys=True):
309		"""Override the Client.get method and read data into object::
310
311		document = Document(auth)
312		document.get(settings.ROOT_URL + "/api/")
313
314		Args:
315		url (str): url to request
316		force_keys (bool): If True, define a new class attribute from data
317		keys
318
319		Returns:
320		requests.Response: a response object
321		"""
322
323		# call the base method
324		response = super().get(url)
325
326		# read data
327		self.read_data(response.json(), force_keys)
328
329		# act like client object
330		return response
331
332		def read_data(self, data, force_keys=False):
333		"""Read data from a dictionary object and set class attributes
334
335		Args:
336		data (dict): a data dictionary object read with
337		:py:meth:`response.json() <requests.Response.json>`
338		force_keys (bool): If True, define a new class attribute from data
339		keys
340		"""
341
342		# dealing with this type of documents
343		for key in data.keys():
344		if "date" in key.lower() and is_date(data[key]):
345		self.__update_key(key, parse_date(data[key]), force_keys)
346
347		else:
348		self.__update_key(key, data[key], force_keys)
349
350		self.data = data
351
352		def __update_key(self, key, value, force_keys=False):
353		"""Helper function to update keys"""
354
355		if hasattr(self, key):
356		if getattr(self, key) and getattr(self, key) != '':
357		# when I reload data, I do a substitution
358		logger.debug("Found %s -> %s" % (key, getattr(self, key)))
359		logger.debug("Updating %s -> %s" % (key, value))
360
361		else:
362		# don't have this attribute set
363		logger.debug("Setting %s -> %s" % (key, value))
364
365		setattr(self, key, value)
366
367		else:
368		if force_keys is True:
369		logger.debug("Forcing %s -> %s" % (key, value))
370		setattr(self, key, value)
371
372		else:
373		logger.warning("key %s not implemented" % (key))
374
375		@classmethod
376		def clean_url(cls, url):
377		"""Remove stuff like ``{?projection}`` from url
378
379		Args:
380		url (str): a string url
381
382		Returns:
383		str: the cleaned url
384		"""
385
386		# remove {?projection} from self url. This is unreachable
387		if '{?projection}' in url:
388		logger.debug("removing {?projection} from url")
389		url = url.replace("{?projection}", "")
390
391		return url
392
393		@classmethod
394		def read_url(cls, auth, url):
395		"""Read a url and returns a :py:class:`Document` object
396
397		Args:
398		auth (Auth): an Auth object to pass to result
399		url (str): url to request
400
401		Returns:
402		Document: a document object
403		"""
404
405		# clean url
406		url = cls.clean_url(url)
407
408		# create a new document
409		document = cls(auth=auth)
410
411		# get url and load data
412		document.get(url)
413
414		return document
415
416		def paginate(self):
417		"""Follow all the pages. Return an iterator of document objects
418
419		Args:
420		response (requests.Response): a response object
421
422		Yield:
423		Document: a new Document instance
424		"""
425
426		# return myself
427		yield self
428
429		# track the current document
430		document = self
431
432		while 'next' in document._links:
433		url = document._links['next']['href']
434		document = Document.read_url(self.auth, url)
435
436		# return the last document
437		yield document
438
439		def follow_tag(self, tag, force_keys=True):
440		"""Pick a url from data attribute relying on tag, perform a request
441		and returns a document object. For instance::
442
443		document.follow_tag('userSubmissions')
444
445		will return a document instance by requesting with
446		:py:meth:`Client.get` using
447		``document._links['userSubmissions']['href']`` as url
448
449		Args:
450		tag (str): a key from USI response dictionary
451		force_keys (bool): set a new class attribute if not present
452
453		Returns:
454		Document: a document object
455		"""
456
457		logger.debug("Following %s url" % (tag))
458
459		url = self._links[tag]['href']
460
461		# create a new document
462		document = Document(auth=self.auth)
463
464		# read data
465		document.get(url, force_keys)
466
467		return document
468
469		def follow_self_url(self):
470		"""Follow self url and update class attributes. For instance::
471
472		document.follow_self_url()
473
474		will reload document instance by requesting with
475		:py:meth:`Client.get` using
476		``document.data['_links']['self']['href']`` as url"""
477
478		logger.debug("Following self url")
479
480		# get a url to follow
481		url = self._links['self']['href']
482
483		# clean url
484		url = self.clean_url(url)
485
486		logger.debug("Updating self")
487
488		# now follow self url and load data to self
489		self.get(url)
490

cnr-ibba / pyUSIrest

Pull Request — master (#7)

pyUSIrest.client.Submission.get_samples() B

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like