b2blaze.models.file_list

Complexity

Total Complexity

27

Size/Duplication

Total Lines	209
Duplicated Lines	0 %

Test Coverage

Coverage

57.66%

Importance

Changes

0

6 Methods

Rating	Name	Duplication	Size	Complexity
A	B2FileList.all()	0	6	1
A	B2FileList.__init__()	0	10	1
B	B2FileList.get()	0	36	6
B	B2FileList._update_files_list()	0	32	6
C	B2FileList.upload_large_file()	0	68	9
A	B2FileList.upload()	0	30	4

"""
Copyright George Sibble 2018
"""

from ..b2_exceptions import B2InvalidBucketName, B2InvalidBucketConfiguration, B2BucketCreationError

Unused B2InvalidBucketName imported from b2_exceptions

Unused B2BucketCreationError imported from b2_exceptions

Unused B2InvalidBucketConfiguration imported from b2_exceptions

from .b2_file import B2File
from ..utilities import b2_url_encode, get_content_length, get_part_ranges, decode_error, RangeStream, StreamWithHashProgress

This line is too long as per the coding-style (125/100).

Unused StreamWithHashProgress imported from utilities

from ..b2_exceptions import B2RequestError, B2FileNotFound
from multiprocessing.dummy import Pool as ThreadPool

standard import "from multiprocessing.dummy import Pool as ThreadPool" should be placed before "from ..b2_exceptions import B2InvalidBucketName, B2InvalidBucketConfiguration, B2BucketCreationError"

class B2FileList(object):

Empty class docstring

    """

    """
    def __init__(self, connector, bucket):
        """

        :param connector:
        :param bucket:
        """
        self.connector = connector
        self.bucket = bucket
        self._files_by_name = {}
        self._files_by_id = {}

    def all(self):
        """

        :return:
        """
        return self._update_files_list(retrieve=True)

    def _update_files_list(self, retrieve=False):
        """

        :param retrieve:
        :return:
        """
        path = '/b2_list_file_names'
        files = []
        new_files_to_retrieve = True
        params = {
            'bucketId': self.bucket.bucket_id,
            'maxFileCount': 10000
        }
        while new_files_to_retrieve:
            response = self.connector.make_request(path=path, method='post', params=params)
            if response.status_code == 200:
                files_json = response.json()
                self._files_by_name = {}
                self._files_by_id = {}
                for file_json in files_json['files']:
                    new_file = B2File(connector=self.connector, parent_list=self, **file_json)
                    files.append(new_file)
                    self._files_by_name[file_json['fileName']] = new_file
                    self._files_by_id[file_json['fileId']] = new_file
                if files_json['nextFileName'] is None:
                    new_files_to_retrieve = False
                else:
                    params['startFileName'] = files_json['nextFileName']
            else:
                raise B2RequestError(decode_error(response))
        if retrieve:
            return files

    def get(self, file_name=None, file_id=None):
        """

        :param file_name:
        :param file_id:
        :return:
        """
        if file_name is not None:
            path = '/b2_list_file_names'
            params = {
                'prefix': b2_url_encode(file_name),
                'bucketId': self.bucket.bucket_id
            }

            response = self.connector.make_request(path, method='post', params=params)
            if response.status_code == 200:
                file_json = response.json()
                if len(file_json['files']) > 0:

Do not use len(SEQUENCE) to determine if a sequence is empty

                    return B2File(connector=self.connector, parent_list=self, **file_json['files'][0])

This line is too long as per the coding-style (102/100).

                else:
                    raise B2FileNotFound('fileName - ' + file_name)
            else:
                raise B2RequestError(decode_error(response))
        elif file_id is not None:
            path = '/b2_get_file_info'
            params = {
                'fileId': file_id
            }
            response = self.connector.make_request(path, method='post', params=params)
            if response.status_code == 200:
                file_json = response.json()
                return B2File(connector=self.connector, parent_list=self, **file_json)
            else:
                raise B2RequestError(decode_error(response))
        else:
            raise ValueError('file_name or file_id must be passed')

        # self._update_files_list()
        # if file_name is not None:
        #     return self._files_by_name.get(file_name, None)
        # else:
        #     return self._files_by_id.get(file_id, None)
        # pass

    def upload(self, contents, file_name, mime_content_type=None, content_length=None, progress_listener=None):

This line is too long as per the coding-style (111/100).

Too many arguments (6/5)

The argument mime_content_type seems to be unused.

        """

        :param contents:
        :param file_name:
        :param mime_content_type:
        :param content_length:
        :param progress_listener:
        :return:
        """
        if file_name[0] == '/':
            file_name = file_name[1:]
        get_upload_url_path = '/b2_get_upload_url'
        params = {
            'bucketId': self.bucket.bucket_id
        }
        upload_url_response = self.connector.make_request(path=get_upload_url_path, method='post', params=params)

This line is too long as per the coding-style (113/100).

        if upload_url_response.status_code == 200:
            upload_url = upload_url_response.json().get('uploadUrl', None)
            auth_token = upload_url_response.json().get('authorizationToken', None)
            upload_response = self.connector.upload_file(file_contents=contents, file_name=file_name,

This line is too long as per the coding-style (101/100).

                                                         upload_url=upload_url, auth_token=auth_token,

This line is too long as per the coding-style (102/100).

                                                         content_length=content_length, progress_listener=progress_listener)

This line is too long as per the coding-style (124/100).

            if upload_response.status_code == 200:
                new_file = B2File(connector=self.connector, parent_list=self, **upload_response.json())

This line is too long as per the coding-style (103/100).

                return new_file
            else:
                raise B2RequestError(decode_error(upload_response))
        else:
            raise B2RequestError(decode_error(upload_url_response))

    def upload_large_file(self, contents, file_name, part_size=None, num_threads=4,

Too many arguments (8/5)

This function exceeds the maximum number of variables (19/15).

                          mime_content_type=None, content_length=None, progress_listener=None):
        """

        :param contents:
        :param file_name:
        :param part_size:
        :param num_threads:
        :param mime_content_type:
        :param content_length:
        :param progress_listener:
        :return:
        """
        if file_name[0] == '/':
            file_name = file_name[1:]
        if part_size == None:

Comparison to None should be 'expr is None'

            part_size = self.connector.recommended_part_size
        if content_length == None:

Comparison to None should be 'expr is None'

            content_length = get_content_length(contents)
        start_large_file_path = '/b2_start_large_file'
        params = {
            'bucketId': self.bucket.bucket_id,
            'fileName': b2_url_encode(file_name),
            'contentType': mime_content_type or 'b2/x-auto'
        }
        large_file_response = self.connector.make_request(path=start_large_file_path, method='post', params=params)

This line is too long as per the coding-style (115/100).

        if large_file_response.status_code == 200:
            file_id = large_file_response.json().get('fileId', None)
            get_upload_part_url_path = '/b2_get_upload_part_url'
            params = {
                'fileId': file_id
            }
            pool = ThreadPool(num_threads)
            def upload_part_worker(args):

This function should have a docstring.

                part_number, part_range = args
                offset, content_length = part_range
                with open(contents.name, 'rb') as file:

This seems to re-define the built-in file.

                    file.seek(offset)
                    stream = RangeStream(file, offset, content_length)
                    upload_part_url_response = self.connector.make_request(path=get_upload_part_url_path, method='post', params=params)

This line is too long as per the coding-style (135/100).

                    if upload_part_url_response.status_code == 200:
                        upload_url = upload_part_url_response.json().get('uploadUrl')
                        auth_token = upload_part_url_response.json().get('authorizationToken')
                        upload_part_response = self.connector.upload_part(file_contents=stream, content_length=content_length,

This line is too long as per the coding-style (126/100).

                                                                          part_number=part_number, upload_url=upload_url,

This line is too long as per the coding-style (121/100).

                                                                          auth_token=auth_token, progress_listener=progress_listener)

This line is too long as per the coding-style (133/100).

                        if upload_part_response.status_code == 200:
                            return upload_part_response.json().get('contentSha1', None)
                        else:
                            raise B2RequestError(decode_error(upload_part_response))
                    else:
                        raise B2RequestError(decode_error(upload_part_url_response))
            sha_list = pool.map(upload_part_worker, enumerate(get_part_ranges(content_length, part_size), 1))

This line is too long as per the coding-style (109/100).

            pool.close()
            pool.join()
            finish_large_file_path = '/b2_finish_large_file'
            params = {
                'fileId': file_id,
                'partSha1Array': sha_list
            }
            finish_large_file_response = self.connector.make_request(path=finish_large_file_path, method='post', params=params)

This line is too long as per the coding-style (127/100).

            if finish_large_file_response.status_code == 200:
                new_file = B2File(connector=self.connector, parent_list=self, **finish_large_file_response.json())

This line is too long as per the coding-style (114/100).

                return new_file
            else:
                raise B2RequestError(decode_error(finish_large_file_response))
        else:
            raise B2RequestError(decode_error(large_file_response))