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-23 06:55 UTC

b2blaze.connector.B2Connector.upload_part() A

↳ Parent: b2blaze.connector

Complexity

Conditions

Size

Total Lines	23
Code Lines	10

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	1.5786

Importance

Changes

Metric	Value
cc	1
eloc	10
nop	7
dl	0
loc	23
ccs	1
cts	6
cp	0.1666
crap	1.5786
rs	9.9
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, decode_error, get_content_length, StreamWithHashProgress


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 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 content_length is None:
            if hasattr(file_contents, 'read'):
                file_sha = 'hex_digits_at_end'
                content_length = get_content_length(file_contents)
                data = StreamWithHashProgress(stream=file_contents, progress_listener=progress_listener)

                content_length += data.hash_size()
            else:
                file_sha = sha1(file_contents).hexdigest()
                content_length = len(file_contents)
                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
        }

        result = requests.post(upload_url, headers=headers, data=data)


        return result

    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.post(upload_url, headers=headers, data=data)

    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-07-23 06:58 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, decode_error, get_content_length, StreamWithHashProgress
		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 (101/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
11
12	1	class B2Connector(object):
		0 ignored issues – show Documentation introduced 2018-07-23 06:58 UTC by Report Bug Copy Issue Report Empty class docstring Loading history... best-practice introduced 2018-07-23 06:58 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 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...
104		direct=False, mime_content_type=None, content_length=None, 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 (103/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history... Unused Code introduced 2018-07-23 06:58 UTC by Report Bug Copy Issue Report The argument `direct` seems to be unused. Loading history...
105		"""
106
107		:param file_contents:
108		:param file_name:
109		:param upload_url:
110		:param auth_token:
111		:param mime_content_type:
112		:param content_length
113		:param progress_listener
114		:return:
115		"""
116
117	1	if content_length is None:
118	1	if hasattr(file_contents, 'read'):
119	1	file_sha = 'hex_digits_at_end'
120	1	content_length = get_content_length(file_contents)
121	1	data = StreamWithHashProgress(stream=file_contents, progress_listener=progress_listener)
		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 (104/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
122	1	content_length += data.hash_size()
123		else:
124	1	file_sha = sha1(file_contents).hexdigest()
125	1	content_length = len(file_contents)
126	1	data = file_contents
127
128	1	headers = {
129		'Content-Type': mime_content_type or 'b2/x-auto',
130		'Content-Length': str(content_length),
131		'X-Bz-Content-Sha1': file_sha,
		0 ignored issues – show introduced 2018-07-23 06:58 UTC by Report Bug Copy Issue Report The variable `file_sha` does not seem to be defined in case `content_length is None` on line `117` is `False`. Are you sure this can never be the case? Loading history...
132		'X-Bz-File-Name': b2_url_encode(file_name),
133		'Authorization': auth_token
134		}
135
136	1	result = requests.post(upload_url, headers=headers, data=data)
		0 ignored issues – show introduced 2018-07-23 06:58 UTC by Report Bug Copy Issue Report The variable `data` does not seem to be defined in case `content_length is None` on line `117` is `False`. Are you sure this can never be the case? Loading history...
137
138	1	return result
139
140	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...
141		"""
142
143		:param file_contents:
144		:param content_length:
145		:param part_number:
146		:param upload_url:
147		:param auth_token:
148		:param progress_listener:
149		:return:
150		"""
151		file_sha = 'hex_digits_at_end'
152		data = StreamWithHashProgress(stream=file_contents, progress_listener=progress_listener)
153		content_length += data.hash_size()
154
155		headers = {
156		'Content-Length': str(content_length),
157		'X-Bz-Content-Sha1': file_sha,
158		'X-Bz-Part-Number': str(part_number),
159		'Authorization': auth_token
160		}
161
162		return requests.post(upload_url, headers=headers, data=data)
163
164	1	def download_file(self, file_id):
165		"""
166
167		:param file_id:
168		:return:
169		"""
170	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...
171	1	params = {
172		'fileId': file_id
173		}
174	1	headers = {
175		'Authorization': self.auth_token
176		}
177
178	1	return requests.get(download_by_id_url, headers=headers, params=params)
179
		0 ignored issues – show coding-style introduced 2018-06-27 20:20 UTC by Report Bug Copy Issue Report Trailing newlines Loading history...
180

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