b2blaze.connector.B2Connector.upload_part() - Code Metrics - Inspection of "Add Multithreaded Large File Upload" - sibblegp/b2blaze - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#14)

unknown

created 2018-07-20 21:28 UTC

b2blaze.connector.B2Connector.upload_part() A

↳ Parent: b2blaze.connector

Complexity

Conditions

Size

Total Lines	20
Code Lines	8

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	1.4218

Importance

Changes

Metric	Value
cc	1
eloc	8
nop	6
dl	0
loc	20
ccs	1
cts	4
cp	0.25
crap	1.4218
rs	10
c	0
b	0
f	0

"""
Copyright George Sibble 2018
"""
import requests
import datetime

from requests.auth import HTTPBasicAuth
from b2blaze.b2_exceptions import B2AuthorizationError, B2RequestError, B2InvalidRequestType

import sys

from hashlib import sha1

from b2blaze.utilities import b2_url_encode, read_in_chunks, decode_error

class B2Connector(object):

    """

    """
    auth_url = 'https://api.backblazeb2.com/b2api/v1'

    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
        self.api_session = 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()
            else:
                return True


    def _authorize(self):
        """

        :return:
        """
        path = self.auth_url + '/b2_authorize_account'
        result = requests.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'] + '/b2api/v1'
            self.download_url = result_json['downloadUrl'] + '/file/'
            self.recommended_part_size = result_json['recommendedPartSize']
            self.api_session = requests.Session()
            self.api_session.headers.update({
                'Authorization': self.auth_token
            })
        else:
            raise B2AuthorizationError(decode_error(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:
        """
        if self.authorized:
            url = self.api_url + path
            if method == 'get':
                return self.api_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 self.api_session.post(url, json=params, headers=headers)
            else:
                raise B2InvalidRequestType('Request type must be get or post')
        else:
            raise B2AuthorizationError('Unknown Error')

    def _sha(self, file_contents):
class Foo:
    def some_method(self, x, y):
        return x + y;
        buf_size = 65536
        if hasattr(file_contents, 'read'):  # If it has `read`, then it's a file-like object that we can stream

            sha = sha1()
            for chunk in read_in_chunks(file_contents, buf_size):
                sha.update(chunk)
            file_sha = sha.hexdigest()
            file_contents.seek(0)
        else:
            if isinstance(file_contents, str) == True:

                try:
                    file_sha = sha1(file_contents.encode('utf-8')).hexdigest()
                except UnicodeDecodeError:
                    file_sha = sha1(file_contents).hexdigest()
            else:
                file_sha = sha1(file_contents).hexdigest()
        return file_sha

    def upload_file(self, file_contents, file_name, upload_url, auth_token, direct=False, mime_content_type=None):

        """

        :param file_contents:
        :param file_name:
        :param upload_url:
        :param auth_token:
        :param mime_content_type:
        :return:
        """
        file_sha = self._sha(file_contents)

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

        return requests.post(upload_url, headers=headers, data=file_contents)

    def upload_part(self, file_contents, content_length, part_number, upload_url, auth_token):

        """

        :param file_contents:
        :param content_length:
        :param part_number:
        :param upload_url:
        :param auth_token:
        :return:
        """
        file_sha = self._sha(file_contents)

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

        return (requests.post(upload_url, headers=headers, data=file_contents), file_sha)

    def download_file(self, file_id):
        """

        :param file_id:
        :return:
        """
        download_by_id_url = self.download_url.split('file/')[0] + '/b2api/v1/b2_download_file_by_id'

        params = {
            'fileId': file_id
        }
        headers = {
            'Authorization': self.auth_token
        }

        return requests.get(download_by_id_url, headers=headers, params=params)




1		"""
2		Copyright George Sibble 2018
3		"""
4	1	import requests
5	1	import datetime
		0 ignored issues – show introduced 2018-06-27 20:20 UTC by Report Bug Copy Issue Report standard import "import datetime" should be placed before "import requests" Loading history...
6	1	from requests.auth import HTTPBasicAuth
7	1	from b2blaze.b2_exceptions import B2AuthorizationError, B2RequestError, B2InvalidRequestType
		0 ignored issues – show Unused Code introduced 2018-06-27 20:20 UTC by Report Bug Copy Issue Report Unused B2RequestError imported from b2blaze.b2_exceptions Loading history...
8	1	import sys
		0 ignored issues – show Unused Code introduced 2018-06-27 20:20 UTC by Report Bug Copy Issue Report The import `sys` seems to be unused. Loading history... introduced 2018-06-27 20:20 UTC by Report Bug Copy Issue Report standard import "import sys" should be placed before "import requests" Loading history...
9	1	from hashlib import sha1
		0 ignored issues – show introduced 2018-06-27 20:20 UTC by Report Bug Copy Issue Report standard import "from hashlib import sha1" should be placed before "import requests" Loading history...
10	1	from b2blaze.utilities import b2_url_encode, read_in_chunks, decode_error
11
12	1	class B2Connector(object):
		0 ignored issues – show Documentation introduced 2018-06-27 20:20 UTC by Report Bug Copy Issue Report Empty class docstring Loading history... best-practice introduced 2018-07-20 21:15 UTC by Report Bug Copy Issue Report Too many instance attributes (9/7) Loading history...
13		"""
14
15		"""
16	1	auth_url = 'https://api.backblazeb2.com/b2api/v1'
17
18	1	def __init__(self, key_id, application_key):
19		"""
20
21		:param key_id:
22		:param application_key:
23		"""
24	1	self.key_id = key_id
25	1	self.application_key = application_key
26	1	self.account_id = None
27	1	self.auth_token = None
28	1	self.authorized_at = None
29	1	self.api_url = None
30	1	self.download_url = None
31	1	self.recommended_part_size = None
32	1	self.api_session = None
33		#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...
34	1	self._authorize()
35
36
37	1	@property
38		def authorized(self):
		0 ignored issues – show Unused Code introduced 2018-06-27 20:20 UTC by Report Bug Copy Issue Report Either all return statements in a function should return an expression, or none of them should. Loading history...
39		"""
40
41		:return:
42		"""
43	1	if self.auth_token is None:
44		return False
45		else:
46	1	if (datetime.datetime.utcnow() - self.authorized_at) > datetime.timedelta(hours=23):
47		self._authorize()
48		else:
49	1	return True
50
51
52	1	def _authorize(self):
53		"""
54
55		:return:
56		"""
57	1	path = self.auth_url + '/b2_authorize_account'
58	1	result = requests.get(path, auth=HTTPBasicAuth(self.key_id, self.application_key))
59	1	if result.status_code == 200:
60	1	result_json = result.json()
61	1	self.authorized_at = datetime.datetime.utcnow()
62	1	self.account_id = result_json['accountId']
63	1	self.auth_token = result_json['authorizationToken']
64	1	self.api_url = result_json['apiUrl'] + '/b2api/v1'
65	1	self.download_url = result_json['downloadUrl'] + '/file/'
66	1	self.recommended_part_size = result_json['recommendedPartSize']
67	1	self.api_session = requests.Session()
68	1	self.api_session.headers.update({
69		'Authorization': self.auth_token
70		})
71		else:
72		raise B2AuthorizationError(decode_error(result))
73
74
75	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...
76		"""
77
78		:param path:
79		:param method:
80		:param headers:
81		:param params:
82		:param account_id_required:
83		:return:
84		"""
85	1	if self.authorized:
86	1	url = self.api_url + path
87	1	if method == 'get':
88		return self.api_session.get(url, headers=headers)
89	1	elif method == 'post':
90	1	if account_id_required:
91	1	params.update({
92		'accountId': self.account_id
93		})
94	1	headers.update({
95		'Content-Type': 'application/json'
96		})
97	1	return self.api_session.post(url, json=params, headers=headers)
98		else:
99		raise B2InvalidRequestType('Request type must be get or post')
100		else:
101		raise B2AuthorizationError('Unknown Error')
102
103	1	def _sha(self, file_contents):
		0 ignored issues – show Coding Style introduced 2018-07-20 21:15 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...
104	1	buf_size = 65536
105	1	if hasattr(file_contents, 'read'): # If it has `read`, then it's a file-like object that we can stream
		0 ignored issues – show Coding Style introduced 2018-06-27 20:20 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (111/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
106	1	sha = sha1()
107	1	for chunk in read_in_chunks(file_contents, buf_size):
108	1	sha.update(chunk)
109	1	file_sha = sha.hexdigest()
110	1	file_contents.seek(0)
111		else:
112	1	if isinstance(file_contents, str) == True:
		0 ignored issues – show introduced 2018-06-27 20:20 UTC by Report Bug Copy Issue Report Comparison to True should be just 'expr' or 'expr is True' Loading history...
113	1	try:
114	1	file_sha = sha1(file_contents.encode('utf-8')).hexdigest()
115	1	except UnicodeDecodeError:
116	1	file_sha = sha1(file_contents).hexdigest()
117		else:
118		file_sha = sha1(file_contents).hexdigest()
119	1	return file_sha
120
121	1	def upload_file(self, file_contents, file_name, upload_url, auth_token, direct=False, mime_content_type=None):
		0 ignored issues – show Coding Style introduced 2018-06-27 20:20 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (114/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history... best-practice introduced 2018-06-27 20:20 UTC by Report Bug Copy Issue Report Too many arguments (7/5) Loading history... Unused Code introduced 2018-06-27 20:20 UTC by Report Bug Copy Issue Report The argument `direct` seems to be unused. Loading history...
122		"""
123
124		:param file_contents:
125		:param file_name:
126		:param upload_url:
127		:param auth_token:
128		:param mime_content_type:
129		:return:
130		"""
131	1	file_sha = self._sha(file_contents)
132
133	1	headers = {
134		'Content-Type': mime_content_type or 'b2/x-auto',
135		'X-Bz-Content-Sha1': file_sha,
136		'X-Bz-File-Name': b2_url_encode(file_name),
137		'Authorization': auth_token
138		}
139
140	1	return requests.post(upload_url, headers=headers, data=file_contents)
141
142	1	def upload_part(self, file_contents, content_length, part_number, upload_url, auth_token):
		0 ignored issues – show best-practice introduced 2018-07-20 21:15 UTC by Report Bug Copy Issue Report Too many arguments (6/5) Loading history...
143		"""
144
145		:param file_contents:
146		:param content_length:
147		:param part_number:
148		:param upload_url:
149		:param auth_token:
150		:return:
151		"""
152		file_sha = self._sha(file_contents)
153
154		headers = {
155		'Content-Length': str(content_length),
156		'X-Bz-Content-Sha1': file_sha,
157		'X-Bz-Part-Number': str(part_number),
158		'Authorization': auth_token
159		}
160
161		return (requests.post(upload_url, headers=headers, data=file_contents), file_sha)
162
163	1	def download_file(self, file_id):
164		"""
165
166		:param file_id:
167		:return:
168		"""
169	1	download_by_id_url = self.download_url.split('file/')[0] + '/b2api/v1/b2_download_file_by_id'
		0 ignored issues – show Coding Style introduced 2018-06-27 20:20 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (101/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
170	1	params = {
171		'fileId': file_id
172		}
173	1	headers = {
174		'Authorization': self.auth_token
175		}
176
177	1	return requests.get(download_by_id_url, headers=headers, params=params)
178
		0 ignored issues – show coding-style introduced 2018-06-27 20:20 UTC by Report Bug Copy Issue Report Trailing newlines Loading history...
179

sibblegp / b2blaze

GitHub Access Token became invalid

Pull Request — master (#14)

b2blaze.connector.B2Connector.upload_part() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like