b2blaze.connector.requests_retry_session() - Code Metrics - Inspection of "Fix: Fixes connection aborted issue." - sibblegp/b2blaze - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#29)

unknown

created 2019-02-13 17:49 UTC

b2blaze.connector.requests_retry_session() A

↳ Parent: b2blaze.connector

Complexity

Conditions

Size

Total Lines	18
Code Lines	16

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	2

Importance

Changes

Metric	Value
cc	1
eloc	16
nop	4
dl	0
loc	18
ccs	0
cts	7
cp	0
crap	2
rs	9.6
c	0
b	0
f	0

"""
Copyright George Sibble 2018
"""

import datetime
from hashlib import sha1

import requests
from requests.auth import HTTPBasicAuth
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry


import requests_cache


from b2blaze.b2_exceptions import (
    B2Exception,
    B2AuthorizationError,
    B2InvalidRequestType,
)

from b2blaze.utilities import b2_url_encode, get_content_length, StreamWithHashProgress

from .api import BASE_URL, API_VERSION, API


# TODO: Fix this quick hack

requests_cache.install_cache("backblaze_cache", backend="redis", expire_after=180)


def requests_retry_session(
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
    retries=3,

    backoff_factor=0.3,

    status_forcelist=(408, 500, 501, 502, 503, 504),

    session=None,

):
    session = session or requests.Session()
    retry = Retry(
        total=retries,
        read=retries,
        connect=retries,
        backoff_factor=backoff_factor,
        status_forcelist=status_forcelist,
    )
    adapter = HTTPAdapter(max_retries=retry)
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    return session


class B2Connector(object):

    """

    """

    def __init__(self, key_id, application_key):
        """

        :param key_id:
        :param application_key:
        """
        self.key_id = key_id
        self.application_key = application_key
        self.account_id = None
        self.auth_token = None
        self.authorized_at = None
        self.api_url = None
        self.download_url = None
        self.recommended_part_size = None
        # TODO:  Part Size

        self._authorize()

    @property
    def authorized(self):
        """

        :return:
        """
        if self.auth_token is None:
            return False
        else:
            if (datetime.datetime.utcnow() - self.authorized_at) > datetime.timedelta(
                hours=23

            ):
                self._authorize()
            return True

    def _authorize(self):
        """

        :return:
        """
        path = BASE_URL + API.authorize

        result = requests_retry_session().get(
            path, auth=HTTPBasicAuth(self.key_id, self.application_key)
        )
        if result.status_code == 200:
            result_json = result.json()
            self.authorized_at = datetime.datetime.utcnow()
            self.account_id = result_json["accountId"]
            self.auth_token = result_json["authorizationToken"]
            self.api_url = result_json["apiUrl"] + API_VERSION
            self.download_url = (
                result_json["downloadUrl"] + API_VERSION + API.download_file_by_id
            )
            self.recommended_part_size = result_json["recommendedPartSize"]
        else:
            raise B2Exception.parse(result)

    def make_request(
# Bad:
# If array_param is modified inside the function, the next invocation will
# receive the modified object.
def some_function(array_param=[]):
    # ...

# Better: Create an array on each invocation
def some_function(array_param=None):
    array_param = array_param or []
    # ...
        self, path, method="get", headers={}, params={}, account_id_required=False

    ):
        """

        :param path:
        :param method:
        :param headers:
        :param params:
        :param account_id_required:
        :return:
        """
        headers.update({"Authorization": self.auth_token})

        if self.authorized:
            url = self.api_url + path
            if method == "get":
                return requests_retry_session().get(url, headers=headers)
            elif method == "post":
                if account_id_required:
                    params.update({"accountId": self.account_id})
                headers.update({"Content-Type": "application/json"})
                return requests_retry_session().post(url, json=params, headers=headers)
            else:
                raise B2InvalidRequestType("Request type must be get or post")
        else:
            raise B2AuthorizationError("Unknown Error")

    def upload_file(
class Foo:
    def some_method(self, x, y):
        return x + y;
        self,

        file_contents,

        file_name,

        upload_url,

        auth_token,

        direct=False,

        mime_content_type=None,

        content_length=None,

        progress_listener=None,

    ):
        """

        :param file_contents:
        :param file_name:
        :param upload_url:
        :param auth_token:
        :param mime_content_type:
        :param content_length
        :param progress_listener
        :return:
        """
        if hasattr(file_contents, "read"):
            if content_length is None:
                content_length = get_content_length(file_contents)
            file_sha = "hex_digits_at_end"
            data = StreamWithHashProgress(
                stream=file_contents, progress_listener=progress_listener
            )
            content_length += data.hash_size()
        else:
            if content_length is None:
                content_length = len(file_contents)
            file_sha = sha1(file_contents).hexdigest()
            data = file_contents

        headers = {
            "Content-Type": mime_content_type or "b2/x-auto",
            "Content-Length": str(content_length),
            "X-Bz-Content-Sha1": file_sha,
            "X-Bz-File-Name": b2_url_encode(file_name),
            "Authorization": auth_token,
        }

        return requests_retry_session().post(upload_url, headers=headers, data=data)

    def upload_part(
class Foo:
    def some_method(self, x, y):
        return x + y;
        self,

        file_contents,

        content_length,

        part_number,

        upload_url,

        auth_token,

        progress_listener=None,

    ):
        """

        :param file_contents:
        :param content_length:
        :param part_number:
        :param upload_url:
        :param auth_token:
        :param progress_listener:
        :return:
        """
        file_sha = "hex_digits_at_end"
        data = StreamWithHashProgress(
            stream=file_contents, progress_listener=progress_listener
        )
        content_length += data.hash_size()

        headers = {
            "Content-Length": str(content_length),
            "X-Bz-Content-Sha1": file_sha,
            "X-Bz-Part-Number": str(part_number),
            "Authorization": auth_token,
        }

        return requests_retry_session().post(upload_url, headers=headers, data=data)

    def download_file(self, file_id):
        """

        :param file_id:
        :return:
        """
        url = self.download_url
        params = {"fileId": file_id}
        headers = {"Authorization": self.auth_token}

        return requests_retry_session().get(url, headers=headers, params=params)


1		"""
2		Copyright George Sibble 2018
3		"""
4
5	1	import datetime
6	1	from hashlib import sha1
7
8	1	import requests
9	1	from requests.auth import HTTPBasicAuth
10	1	from requests.adapters import HTTPAdapter
11	1	from requests.packages.urllib3.util.retry import Retry
		0 ignored issues – show introduced 2019-02-13 17:44 UTC by Report Bug Copy Issue Report Unable to import 'requests.packages.urllib3.util.retry' Loading history...
12
13	1	import requests_cache
		0 ignored issues – show introduced 2019-02-13 17:44 UTC by Report Bug Copy Issue Report Unable to import 'requests_cache' Loading history...
14
15		from b2blaze.b2_exceptions import (
16		B2Exception,
17		B2AuthorizationError,
18		B2InvalidRequestType,
19		)
20
21		from b2blaze.utilities import b2_url_encode, get_content_length, StreamWithHashProgress
22
23		from .api import BASE_URL, API_VERSION, API
24
25
26		# TODO: Fix this quick hack
		0 ignored issues – show Coding Style introduced 2019-02-13 17:44 UTC by Report Bug Copy Issue Report `TODO` and `FIXME` comments should generally be avoided. Loading history...
27		requests_cache.install_cache("backblaze_cache", backend="redis", expire_after=180)
28
29
30		def requests_retry_session(
		0 ignored issues – show Coding Style introduced 2019-02-13 17:44 UTC by Report Bug Copy Issue Report This function should have a docstring. The coding style of this project requires that you add a docstring to this code element. Below, you find an example for methods: class SomeClass: def some_method(self): """Do x and return foo.""" If you would like to know more about docstrings, we recommend to read PEP-257: Docstring Conventions. Loading history...
31		retries=3,
		0 ignored issues – show Coding Style introduced 2019-02-13 17:44 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
32		backoff_factor=0.3,
		0 ignored issues – show Coding Style introduced 2019-02-13 17:44 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
33		status_forcelist=(408, 500, 501, 502, 503, 504),
		0 ignored issues – show Coding Style introduced 2019-02-13 17:44 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
34		session=None,
		0 ignored issues – show Coding Style introduced 2019-02-13 17:44 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
35		):
36		session = session or requests.Session()
37		retry = Retry(
38		total=retries,
39		read=retries,
40		connect=retries,
41		backoff_factor=backoff_factor,
42		status_forcelist=status_forcelist,
43		)
44		adapter = HTTPAdapter(max_retries=retry)
45		session.mount("http://", adapter)
46		session.mount("https://", adapter)
47		return session
48
49
50		class B2Connector(object):
		0 ignored issues – show Documentation introduced 2019-02-13 17:44 UTC by Report Bug Copy Issue Report Empty class docstring Loading history... best-practice introduced 2019-02-13 17:44 UTC by Report Bug Copy Issue Report Too many instance attributes (8/7) Loading history...
51		"""
52
53		"""
54
55		def __init__(self, key_id, application_key):
56		"""
57
58		:param key_id:
59		:param application_key:
60		"""
61		self.key_id = key_id
62		self.application_key = application_key
63		self.account_id = None
64		self.auth_token = None
65		self.authorized_at = None
66		self.api_url = None
67		self.download_url = None
68		self.recommended_part_size = None
69		# TODO: Part Size
		0 ignored issues – show Coding Style introduced 2018-06-27 20:20 UTC by Report Bug Copy Issue Report `TODO` and `FIXME` comments should generally be avoided. Loading history...
70		self._authorize()
71
72		@property
73		def authorized(self):
74		"""
75
76		:return:
77		"""
78		if self.auth_token is None:
79		return False
80		else:
81		if (datetime.datetime.utcnow() - self.authorized_at) > datetime.timedelta(
82		hours=23
		0 ignored issues – show Coding Style introduced 2019-02-13 17:44 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
83		):
84		self._authorize()
85		return True
86
87		def _authorize(self):
88		"""
89
90		:return:
91		"""
92		path = BASE_URL + API.authorize
93
94		result = requests_retry_session().get(
95		path, auth=HTTPBasicAuth(self.key_id, self.application_key)
96		)
97		if result.status_code == 200:
98		result_json = result.json()
99		self.authorized_at = datetime.datetime.utcnow()
100		self.account_id = result_json["accountId"]
101		self.auth_token = result_json["authorizationToken"]
102		self.api_url = result_json["apiUrl"] + API_VERSION
103		self.download_url = (
104		result_json["downloadUrl"] + API_VERSION + API.download_file_by_id
105		)
106		self.recommended_part_size = result_json["recommendedPartSize"]
107		else:
108		raise B2Exception.parse(result)
109
110		def make_request(
		0 ignored issues – show Bug Best Practice introduced 2019-02-13 17:44 UTC by Report Bug Copy Issue Report The default value `{}` might cause unintended side-effects. Objects as default values are only created once in Python and not on each invocation of the function. If the default object is modified, this modification is carried over to the next invocation of the method. # Bad: # If array_param is modified inside the function, the next invocation will # receive the modified object. def some_function(array_param=[]): # ... # Better: Create an array on each invocation def some_function(array_param=None): array_param = array_param or [] # ... Loading history... best-practice introduced 2019-02-13 17:44 UTC by Report Bug Copy Issue Report Too many arguments (6/5) Loading history...
111		self, path, method="get", headers={}, params={}, account_id_required=False
		0 ignored issues – show Coding Style introduced 2019-02-13 17:44 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
112		):
113		"""
114
115		:param path:
116		:param method:
117		:param headers:
118		:param params:
119		:param account_id_required:
120		:return:
121		"""
122		headers.update({"Authorization": self.auth_token})
123
124		if self.authorized:
125		url = self.api_url + path
126		if method == "get":
127		return requests_retry_session().get(url, headers=headers)
128		elif method == "post":
129		if account_id_required:
130		params.update({"accountId": self.account_id})
131		headers.update({"Content-Type": "application/json"})
132		return requests_retry_session().post(url, json=params, headers=headers)
133		else:
134		raise B2InvalidRequestType("Request type must be get or post")
135		else:
136		raise B2AuthorizationError("Unknown Error")
137
138		def upload_file(
		0 ignored issues – show best-practice introduced 2019-02-13 17:44 UTC by Report Bug Copy Issue Report Too many arguments (9/5) Loading history... Coding Style introduced 2019-02-13 17:44 UTC by Report Bug Copy Issue Report This method could be written as a function/class method. If a method does not access any attributes of the class, it could also be implemented as a function or static method. This can help improve readability. For example class Foo: def some_method(self, x, y): return x + y; could be written as class Foo: @classmethod def some_method(cls, x, y): return x + y; Loading history...
139		self,
		0 ignored issues – show Coding Style introduced 2019-02-13 17:44 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
140		file_contents,
		0 ignored issues – show Coding Style introduced 2019-02-13 17:44 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
141		file_name,
		0 ignored issues – show Coding Style introduced 2019-02-13 17:44 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
142		upload_url,
		0 ignored issues – show Coding Style introduced 2019-02-13 17:44 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
143		auth_token,
		0 ignored issues – show Coding Style introduced 2019-02-13 17:44 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
144		direct=False,
		0 ignored issues – show Coding Style introduced 2019-02-13 17:44 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history... Unused Code introduced 2019-02-13 17:44 UTC by Report Bug Copy Issue Report The argument `direct` seems to be unused. Loading history...
145		mime_content_type=None,
		0 ignored issues – show Coding Style introduced 2019-02-13 17:44 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
146		content_length=None,
		0 ignored issues – show Coding Style introduced 2019-02-13 17:44 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
147		progress_listener=None,
		0 ignored issues – show Coding Style introduced 2019-02-13 17:44 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
148		):
149		"""
150
151		:param file_contents:
152		:param file_name:
153		:param upload_url:
154		:param auth_token:
155		:param mime_content_type:
156		:param content_length
157		:param progress_listener
158		:return:
159		"""
160		if hasattr(file_contents, "read"):
161		if content_length is None:
162		content_length = get_content_length(file_contents)
163		file_sha = "hex_digits_at_end"
164		data = StreamWithHashProgress(
165		stream=file_contents, progress_listener=progress_listener
166		)
167		content_length += data.hash_size()
168		else:
169		if content_length is None:
170		content_length = len(file_contents)
171		file_sha = sha1(file_contents).hexdigest()
172		data = file_contents
173
174		headers = {
175		"Content-Type": mime_content_type or "b2/x-auto",
176		"Content-Length": str(content_length),
177		"X-Bz-Content-Sha1": file_sha,
178		"X-Bz-File-Name": b2_url_encode(file_name),
179		"Authorization": auth_token,
180		}
181
182		return requests_retry_session().post(upload_url, headers=headers, data=data)
183
184		def upload_part(
		0 ignored issues – show best-practice introduced 2019-02-13 17:44 UTC by Report Bug Copy Issue Report Too many arguments (7/5) Loading history... Coding Style introduced 2019-02-13 17:44 UTC by Report Bug Copy Issue Report This method could be written as a function/class method. If a method does not access any attributes of the class, it could also be implemented as a function or static method. This can help improve readability. For example class Foo: def some_method(self, x, y): return x + y; could be written as class Foo: @classmethod def some_method(cls, x, y): return x + y; Loading history...
185		self,
		0 ignored issues – show Coding Style introduced 2019-02-13 17:44 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
186		file_contents,
		0 ignored issues – show Coding Style introduced 2019-02-13 17:44 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
187		content_length,
		0 ignored issues – show Coding Style introduced 2019-02-13 17:44 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
188		part_number,
		0 ignored issues – show Coding Style introduced 2019-02-13 17:44 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
189		upload_url,
		0 ignored issues – show Coding Style introduced 2019-02-13 17:44 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
190		auth_token,
		0 ignored issues – show Coding Style introduced 2019-02-13 17:44 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
191		progress_listener=None,
		0 ignored issues – show Coding Style introduced 2019-02-13 17:44 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
192		):
193		"""
194
195		:param file_contents:
196		:param content_length:
197		:param part_number:
198		:param upload_url:
199		:param auth_token:
200		:param progress_listener:
201		:return:
202		"""
203		file_sha = "hex_digits_at_end"
204		data = StreamWithHashProgress(
205		stream=file_contents, progress_listener=progress_listener
206		)
207		content_length += data.hash_size()
208
209		headers = {
210		"Content-Length": str(content_length),
211		"X-Bz-Content-Sha1": file_sha,
212		"X-Bz-Part-Number": str(part_number),
213		"Authorization": auth_token,
214		}
215
216		return requests_retry_session().post(upload_url, headers=headers, data=data)
217
218		def download_file(self, file_id):
219		"""
220
221		:param file_id:
222		:return:
223		"""
224		url = self.download_url
225		params = {"fileId": file_id}
226		headers = {"Authorization": self.auth_token}
227
228		return requests_retry_session().get(url, headers=headers, params=params)
229

sibblegp / b2blaze

GitHub Access Token became invalid

Pull Request — master (#29)

b2blaze.connector.requests_retry_session() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like