b2blaze.models.file_list

Complexity

Total Complexity

44

Size/Duplication

Total Lines	319
Duplicated Lines	0 %

Test Coverage

Coverage

10.63%

Importance

Changes

0

11 Methods

Rating	Name	Duplication	Size	Complexity
A	B2FileList._get_by_id()	0	12	2
C	B2FileList.upload_large_file()	0	68	9
A	B2FileList._get_by_name()	0	18	3
B	B2FileList.all_file_versions()	0	52	7
A	B2FileList.get_versions()	0	21	3
A	B2FileList.all()	0	20	3
A	B2FileList.__init__()	0	8	1
A	B2FileList.get()	0	18	3
B	B2FileList._retrieve_file_list()	0	26	5
A	B2FileList.upload()	0	30	4
A	B2FileList.delete_all()	0	15	4

"""
Copyright George Sibble 2018
"""

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 decode_error imported from utilities

Unused StreamWithHashProgress imported from utilities

from ..b2_exceptions import B2Exception, B2FileNotFoundError
from multiprocessing.dummy import Pool as ThreadPool

standard import "from multiprocessing.dummy import Pool as ThreadPool" should be placed before "from .b2_file import B2File"

from ..api import API

class B2FileList(object):

Empty class docstring

    """

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

        :param connector:
        :param bucket:
        """
        self.connector = connector
        self.bucket = bucket

    def all(self, include_hidden=False, limit=None):
        """ Return an updated list of all files.
            (This does not include hidden files unless include_hidden flag set to True)

            Parameters:
                include_hidden:         (bool) Include hidden files
                limit:                  (int)  Limit number of file results

        """
        if not include_hidden:
            return self._retrieve_file_list(limit=limit)
        else:
            results = self.all_file_versions(limit=limit)
            versions = results['file_versions']
            file_ids = results['file_ids']
            if versions:
                # Return only the first file from a given file with multiple versions
                files = [versions[f][0] for f in file_ids]
                return files
        return []   # Return empty set on no results

    def delete_all(self, confirm=False):
        """ Delete all files in the bucket. 

Trailing whitespace

            Parameters:
                confirm:    (bool)  Safety check. Confirm deletion
        """ 

Trailing whitespace

        if not confirm:
            raise Exception('This will delete all files! Pass confirm=True')
        

Trailing whitespace

        all_files = self.all(include_hidden=True)
        try:
            for f in all_files:

The name f does not conform to the variable naming conventions ((([a-z][a-z0-9_]{2,30})|(_[a-z0-9_]*))$).

                f.delete_all_versions(confirm=True)
        except Exception as E:

The name E does not conform to the variable naming conventions ((([a-z][a-z0-9_]{2,30})|(_[a-z0-9_]*))$).

            raise B2Exception.parse(E)
        return []

        

Trailing whitespace

    def _retrieve_file_list(self, limit=10000):
        """ Retrieve list of all files in bucket 

Trailing whitespace

            Parameters:
                limit:      (int)  Max number of file results, default 10000
        """
        path = API.list_all_files
        files = []
        new_files_to_retrieve = True
        params = {
            'bucketId': self.bucket.bucket_id,
            'maxFileCount': limit,
        }
        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()
                for file_json in files_json['files']:
                    new_file = B2File(connector=self.connector, parent_list=self, **file_json)
                    files.append(new_file)
                if files_json['nextFileName'] is None:
                    new_files_to_retrieve = False
                else:
                    params['startFileName'] = files_json['nextFileName']
            else:
                raise B2Exception.parse(response)
        return files


    def get(self, file_name=None, file_id=None):
        """ Get a file by file name or id.
            Required:
                file_name or file_id

            Parameters:
                file_name:          (str) File name 

Trailing whitespace

                file_id:            (str) File ID 

Trailing whitespace

        """
        if file_name:
            file = self._get_by_name(file_name)

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

        elif file_id:
            file = self._get_by_id(file_id)
        else:
            raise ValueError('file_name or file_id must be passed')
        

Trailing whitespace

        return file


    def get_versions(self, file_name=None, file_id=None, limit=None):

The argument limit seems to be unused.

        """ Return list of all the versions of one file in current bucket. 

Trailing whitespace

            Required:
                file_id or file_name   (either)

            Params:
                file_id:            (str) File id
                file_name:          (str) File id
                limit:              (int) Limit number of results returned (optional)

            Returns:
                file_versions       (list) B2FileObject of all file versions
        """ 

Trailing whitespace

        if file_name:
            file = self.get(file_name)

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

        elif file_id:
            file = self.get(file_id=file_id)
        else:
            raise ValueError('Either file_id or file_name required for get_versions')
        return file.get_versions()
        

Trailing whitespace

    def all_file_versions(self, limit=None):
        """ Return all the versions of all files in a given bucket.

            Params:
                limit:              (int) Limit number of results returned (optional). Defaults to 10000

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

            Returns dict: 

Trailing whitespace

                'file_names':       (list) String filenames
                'file_ids':         (list) File IDs
                'file_versions':    (dict) b2blaze File objects, keyed by file name
        """ 

Trailing whitespace

        path = API.list_file_versions
        file_versions = dict()
        file_names = []
        file_ids = []
        new_files_to_retrieve = True
        params = {
            'bucketId': self.bucket.bucket_id,
            'maxFileCount': 10000
        }

        # Limit files
        if limit:
            params['maxFileCount'] = limit

        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()
                for file_json in files_json['files']:
                    new_file = B2File(connector=self.connector, parent_list=self, **file_json)

                    # Append file_id, file_name to lists
                    file_name, file_id = file_json['fileName'], file_json['fileId']
                    file_names.append(file_name)
                    file_ids.append(file_id)
                    

Trailing whitespace

                    # Add file to list keyed by file_id
                    if file_id in file_versions:
                        file_versions[file_id].append(new_file)
                    else:
                        file_versions[file_id] = [new_file]

                if files_json['nextFileName'] is None:
                    new_files_to_retrieve = False
                else:
                    params['startFileName'] = files_json['nextFileName']
            else:
                raise B2Exception.parse(response)
        return {'file_names': file_names, 'file_versions': file_versions, 'file_ids': file_ids}


    def _get_by_name(self, file_name):
        """ Internal method, return single file by file name """ 

Trailing whitespace

        path = API.list_all_files
        params = {
            'prefix': b2_url_encode(file_name),
            'bucketId': self.bucket.bucket_id
        }

        response = self.connector.make_request(path, method='post', params=params)
        file_json = response.json()

        # Handle errors and empty files
        if not response.status_code == 200:
            raise B2Exception.parse(response)
        if not len(file_json['files']) > 0:

Do not use len(SEQUENCE) as condition value

            raise B2FileNotFoundError('Filename {} not found'.format(file_name))
        else:
            return B2File(connector=self.connector, parent_list=self, **file_json['files'][0])

    def _get_by_id(self, file_id):
        """ Internal method, return single file by file id """ 

Trailing whitespace

        path = API.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 B2Exception.parse(response)
            

Trailing whitespace

    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 = API.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 B2Exception.parse(upload_response)
        else:
            raise B2Exception.parse(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 = API.upload_large
        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 = API.upload_large_part
            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 B2Exception.parse(upload_part_response)
                    else:
                        raise B2Exception.parse(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 = API.upload_large_finish
            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 B2Exception.parse(finish_large_file_response)
        else:
            raise B2Exception.parse(large_file_response)