b2blaze.connector.B2Connector.upload_file() - Code Metrics - Inspection of "Improve behaviour with restricted application keys" - sibblegp/b2blaze - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#27)

unknown

created 2019-01-15 06:34 UTC

b2blaze.connector.B2Connector.upload_file() A

↳ Parent: b2blaze.connector

Complexity

Conditions

Size

Total Lines	34
Code Lines	19

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	16.5853

Importance

Changes

Metric	Value
cc	4
eloc	19
nop	9
dl	0
loc	34
ccs	1
cts	13
cp	0.0769
crap	16.5853
rs	9.45
c	0
b	0
f	0

How to fix Many Parameters

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

from requests.auth import HTTPBasicAuth
from b2blaze.b2_exceptions import B2Exception, B2AuthorizationError, B2InvalidRequestType
import sys

from hashlib import sha1

from b2blaze.utilities import b2_url_encode, decode_error, get_content_length, StreamWithHashProgress

from .api import BASE_URL, API_VERSION, API

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
        self.api_session = None
        self.bucket_id = 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.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']
            if result_json.get('allowed'):
                self.bucket_id = result_json['allowed'].get('bucketId', None)
            self.api_session = requests.Session()
            self.api_session.headers.update({
                'Authorization': self.auth_token
            })
        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:
        """
        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 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.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.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.get(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 B2Exception, B2AuthorizationError, B2InvalidRequestType
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... Unused Code introduced 2018-10-30 20:37 UTC by Report Bug Copy Issue Report Unused decode_error imported from b2blaze.utilities Loading history...
11	1	from .api import BASE_URL, API_VERSION, API
12
13	1	class B2Connector(object):
		0 ignored issues – show Documentation introduced 2018-10-30 20:37 UTC by Report Bug Copy Issue Report Empty class docstring Loading history... best-practice introduced 2019-01-15 06:35 UTC by Report Bug Copy Issue Report Too many instance attributes (10/7) Loading history...
14		"""
15
16		"""
17	1	def __init__(self, key_id, application_key):
18		"""
19
20		:param key_id:
21		:param application_key:
22		"""
23	1	self.key_id = key_id
24	1	self.application_key = application_key
25	1	self.account_id = None
26	1	self.auth_token = None
27	1	self.authorized_at = None
28	1	self.api_url = None
29	1	self.download_url = None
30	1	self.recommended_part_size = None
31	1	self.api_session = None
32	1	self.bucket_id = 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):
39		"""
40
41		:return:
42		"""
43		if self.auth_token is None:
44		return False
45		else:
46		if (datetime.datetime.utcnow() - self.authorized_at) > datetime.timedelta(hours=23):
47		self._authorize()
48		return True
49
50
51	1	def _authorize(self):
52		"""
53
54		:return:
55		"""
56	1	path = BASE_URL + API.authorize
57
58	1	result = requests.get(path, auth=HTTPBasicAuth(self.key_id, self.application_key))
59	1	if result.status_code == 200:
60		result_json = result.json()
61		self.authorized_at = datetime.datetime.utcnow()
62		self.account_id = result_json['accountId']
63		self.auth_token = result_json['authorizationToken']
64		self.api_url = result_json['apiUrl'] + API_VERSION
65		self.download_url = result_json['downloadUrl'] + API_VERSION + API.download_file_by_id
66		self.recommended_part_size = result_json['recommendedPartSize']
67		if result_json.get('allowed'):
68		self.bucket_id = result_json['allowed'].get('bucketId', None)
69		self.api_session = requests.Session()
70		self.api_session.headers.update({
71		'Authorization': self.auth_token
72		})
73		else:
74	1	raise B2Exception.parse(result)
75
76
77	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...
78		"""
79
80		:param path:
81		:param method:
82		:param headers:
83		:param params:
84		:param account_id_required:
85		:return:
86		"""
87		if self.authorized:
88		url = self.api_url + path
89		if method == 'get':
90		return self.api_session.get(url, headers=headers)
91		elif method == 'post':
92		if account_id_required:
93		params.update({
94		'accountId': self.account_id
95		})
96		headers.update({
97		'Content-Type': 'application/json'
98		})
99		return self.api_session.post(url, json=params, headers=headers)
100		else:
101		raise B2InvalidRequestType('Request type must be get or post')
102		else:
103		raise B2AuthorizationError('Unknown Error')
104
105	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...
106		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...
107		"""
108
109		:param file_contents:
110		:param file_name:
111		:param upload_url:
112		:param auth_token:
113		:param mime_content_type:
114		:param content_length
115		:param progress_listener
116		:return:
117		"""
118		if hasattr(file_contents, 'read'):
119		if content_length is None:
120		content_length = get_content_length(file_contents)
121		file_sha = 'hex_digits_at_end'
122		data = StreamWithHashProgress(stream=file_contents, progress_listener=progress_listener)
123		content_length += data.hash_size()
124		else:
125		if content_length is None:
126		content_length = len(file_contents)
127		file_sha = sha1(file_contents).hexdigest()
128		data = file_contents
129
130		headers = {
131		'Content-Type': mime_content_type or 'b2/x-auto',
132		'Content-Length': str(content_length),
133		'X-Bz-Content-Sha1': file_sha,
134		'X-Bz-File-Name': b2_url_encode(file_name),
135		'Authorization': auth_token
136		}
137
138		return requests.post(upload_url, headers=headers, data=data)
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		url = self.download_url
171		params = {
172		'fileId': file_id
173		}
174		headers = {
175		'Authorization': self.auth_token
176		}
177
178		return requests.get(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 (#27)

b2blaze.connector.B2Connector.upload_file() A

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Many Parameters

Many Parameters

Duplication Side-by-Side

Filter issues like