b2blaze.connector.B2Connector.__init__() - Code Metrics - Inspection of "Fix/Feat: Don't abort connection with session and..." - sibblegp/b2blaze - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#32)

unknown

created 2019-02-15 14:57 UTC

b2blaze.connector.B2Connector.init() A

↳ Parent: b2blaze.connector

Complexity

Conditions

Size

Total Lines	16
Code Lines	10

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	10
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	10
nop	3
dl	0
loc	16
ccs	10
cts	10
cp	1
crap	1
rs	9.9
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


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


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(self, path, method='get', headers={}, params={}, account_id_required=False):
# 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 []
    # ...
        """

        :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(self, file_contents, file_name, upload_url, auth_token,
class Foo:
    def some_method(self, x, y):
        return x + y;
                    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(self, file_contents, content_length, part_number, upload_url, auth_token, progress_listener=None):
class Foo:
    def some_method(self, x, y):
        return x + y;
        """

        :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	from b2blaze.b2_exceptions import (
14		B2Exception,
15		B2AuthorizationError,
16		B2InvalidRequestType,
17		)
18
19	1	from b2blaze.utilities import (
20		b2_url_encode,
21		get_content_length,
22		StreamWithHashProgress,
23		)
24
25	1	from .api import BASE_URL, API_VERSION, API
26
27
28	1	def requests_retry_session(
		0 ignored issues – show Coding Style introduced 2019-02-15 14:59 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...
29		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...
30		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...
31		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...
32		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...
33		):
34	1	session = session or requests.Session()
35	1	retry = Retry(
36		total=retries,
37		read=retries,
38		connect=retries,
39		backoff_factor=backoff_factor,
40		status_forcelist=status_forcelist,
41		)
42	1	adapter = HTTPAdapter(max_retries=retry)
43	1	session.mount('http://', adapter)
44	1	session.mount('https://', adapter)
45	1	return session
46
47
48	1	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...
49		"""
50
51		"""
52	1	def __init__(self, key_id, application_key):
53		"""
54
55		:param key_id:
56		:param application_key:
57		"""
58	1	self.key_id = key_id
59	1	self.application_key = application_key
60	1	self.account_id = None
61	1	self.auth_token = None
62	1	self.authorized_at = None
63	1	self.api_url = None
64	1	self.download_url = None
65	1	self.recommended_part_size = None
66		#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...
67	1	self._authorize()
68
69
70	1	@property
71		def authorized(self):
72		"""
73
74		:return:
75		"""
76		if self.auth_token is None:
77		return False
78		else:
79		if (datetime.datetime.utcnow() - self.authorized_at) > datetime.timedelta(hours=23):
80		self._authorize()
81		return True
82
83
84	1	def _authorize(self):
85		"""
86
87		:return:
88		"""
89	1	path = BASE_URL + API.authorize
90
91	1	result = requests_retry_session().get(
92		path,
93		auth=HTTPBasicAuth(self.key_id, self.application_key)
94		)
95	1	if result.status_code == 200:
96		result_json = result.json()
97		self.authorized_at = datetime.datetime.utcnow()
98		self.account_id = result_json['accountId']
99		self.auth_token = result_json['authorizationToken']
100		self.api_url = result_json['apiUrl'] + API_VERSION
101		self.download_url = result_json['downloadUrl'] + API_VERSION + API.download_file_by_id
102		self.recommended_part_size = result_json['recommendedPartSize']
103		else:
104	1	raise B2Exception.parse(result)
105
106
107	1	def make_request(self, path, method='get', headers={}, params={}, account_id_required=False):
		0 ignored issues – show Bug Best Practice introduced 2018-06-27 20:20 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 2018-06-27 20:20 UTC by Report Bug Copy Issue Report Too many arguments (6/5) Loading history...
108		"""
109
110		:param path:
111		:param method:
112		:param headers:
113		:param params:
114		:param account_id_required:
115		:return:
116		"""
117		headers.update({'Authorization': self.auth_token})
118
119		if self.authorized:
120		url = self.api_url + path
121		if method == 'get':
122		return requests_retry_session().get(url, headers=headers)
123		elif method == 'post':
124		if account_id_required:
125		params.update({
126		'accountId': self.account_id
127		})
128		headers.update({
129		'Content-Type': 'application/json'
130		})
131		return requests_retry_session().post(
132		url,
133		json=params,
134		headers=headers
135		)
136		else:
137		raise B2InvalidRequestType('Request type must be get or post')
138		else:
139		raise B2AuthorizationError('Unknown Error')
140
141	1	def upload_file(self, file_contents, file_name, upload_url, auth_token,
		0 ignored issues – show best-practice introduced 2018-07-23 06:58 UTC by Report Bug Copy Issue Report Too many arguments (9/5) Loading history... Coding Style introduced 2018-07-23 06:58 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...
142		direct=False, mime_content_type=None, content_length=None,
		0 ignored issues – show Unused Code introduced 2019-02-15 14:59 UTC by Report Bug Copy Issue Report The argument `direct` seems to be unused. Loading history...
143		progress_listener=None):
144		"""
145
146		:param file_contents:
147		:param file_name:
148		:param upload_url:
149		:param auth_token:
150		:param mime_content_type:
151		:param content_length
152		:param progress_listener
153		:return:
154		"""
155		if hasattr(file_contents, 'read'):
156		if content_length is None:
157		content_length = get_content_length(file_contents)
158		file_sha = 'hex_digits_at_end'
159		data = StreamWithHashProgress(stream=file_contents, progress_listener=progress_listener)
160		content_length += data.hash_size()
161		else:
162		if content_length is None:
163		content_length = len(file_contents)
164		file_sha = sha1(file_contents).hexdigest()
165		data = file_contents
166
167		headers = {
168		'Content-Type': mime_content_type or 'b2/x-auto',
169		'Content-Length': str(content_length),
170		'X-Bz-Content-Sha1': file_sha,
171		'X-Bz-File-Name': b2_url_encode(file_name),
172		'Authorization': auth_token
173		}
174
175		return requests_retry_session().post(upload_url, headers=headers, data=data)
176
177	1	def upload_part(self, file_contents, content_length, part_number, upload_url, auth_token, progress_listener=None):
		0 ignored issues – show Coding Style introduced 2018-07-23 06:58 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (118/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history... best-practice introduced 2018-07-23 06:58 UTC by Report Bug Copy Issue Report Too many arguments (7/5) Loading history... Coding Style introduced 2018-07-23 06:58 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...
178		"""
179
180		:param file_contents:
181		:param content_length:
182		:param part_number:
183		:param upload_url:
184		:param auth_token:
185		:param progress_listener:
186		:return:
187		"""
188		file_sha = 'hex_digits_at_end'
189		data = StreamWithHashProgress(stream=file_contents, progress_listener=progress_listener)
190		content_length += data.hash_size()
191
192		headers = {
193		'Content-Length': str(content_length),
194		'X-Bz-Content-Sha1': file_sha,
195		'X-Bz-Part-Number': str(part_number),
196		'Authorization': auth_token
197		}
198
199		return requests_retry_session().post(upload_url, headers=headers, data=data)
200
201	1	def download_file(self, file_id):
202		"""
203
204		:param file_id:
205		:return:
206		"""
207		url = self.download_url
208		params = {
209		'fileId': file_id
210		}
211		headers = {
212		'Authorization': self.auth_token
213		}
214
215		return requests_retry_session().get(url, headers=headers, params=params)
216
		0 ignored issues – show coding-style introduced 2018-06-27 20:20 UTC by Report Bug Copy Issue Report Trailing newlines Loading history...
217

sibblegp / b2blaze

GitHub Access Token became invalid

Pull Request — master (#32)

b2blaze.connector.B2Connector.__init__() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

b2blaze.connector.B2Connector.init() A