pyUSIrest.client.Client.post() - Code Metrics - Inspection of ":recycle: Use iterators when browsing USI objects" - cnr-ibba/pyUSIrest - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — devel (#6)

by Paolo

created 2020-01-10 10:38 UTC

pyUSIrest.client.Client.post() A

↳ Parent: pyUSIrest.client

Complexity

Conditions

Size

Total Lines	26
Code Lines	9

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	9
nop	5
dl	0
loc	26
rs	9.95
c	0
b	0
f	0

#!/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

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 RuntimeError("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 ConnectionError(
                "Problems with API endpoints: %s" % response.text)

        # TODO: evaluate a list of expected status?

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

cnr-ibba / pyUSIrest

Pull Request — devel (#6)

pyUSIrest.client.Client.post() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like