b2blaze.models.bucket_list

Complexity

Total Complexity

14

Size/Duplication

Total Lines	97
Duplicated Lines	0 %

Test Coverage

Coverage

91.67%

Importance

Changes

0

5 Methods

Rating	Name	Duplication	Size	Complexity
A	B2Buckets.all()	0	6	1
A	B2Buckets._update_bucket_list()	0	22	4
A	B2Buckets.__init__()	0	8	1
A	B2Buckets.get()	0	12	2
B	B2Buckets.create()	0	28	6

"""
Copyright George Sibble 2018
"""

from ..b2_exceptions import B2Exception, B2InvalidBucketName, B2InvalidBucketConfiguration
from .bucket import B2Bucket
from ..api import API


class B2Buckets(object):

Empty class docstring

    """

    """
    public = 'allPublic'
    private = 'allPrivate'

    def __init__(self, connector):
        """

        :param connector:
        """
        self.connector = connector
        self._buckets_by_name = {}
        self._buckets_by_id = {}

    def all(self):
        """

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

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

        :param retrieve:
        :return:
        """
        path = API.list_all_buckets
        response = self.connector.make_request(path=path, method='post', account_id_required=True)
        if response.status_code == 200:
            response_json = response.json()
            buckets = []
            self._buckets_by_name = {}
            self._buckets_by_id = {}
            for bucket_json in response_json['buckets']:
                new_bucket = B2Bucket(connector=self.connector, parent_list=self, **bucket_json)
                buckets.append(new_bucket)
                self._buckets_by_name[bucket_json['bucketName']] = new_bucket
                self._buckets_by_id[bucket_json['bucketId']] = new_bucket
            if retrieve:
                return buckets
        else:
            raise B2Exception.parse(response)

    def get(self, bucket_name=None, bucket_id=None):
        """

        :param bucket_name:
        :param bucket_id:
        :return:
        """
        self._update_bucket_list()
        if bucket_name is not None:

Unnecessary "else" after "return"

            return self._buckets_by_name.get(bucket_name, None)
        else:
            return self._buckets_by_id.get(bucket_id, None)

    def create(self, bucket_name, security, configuration=None):
        """

        :param bucket_name:
        :param configuration:
        :return:
        """
        path = API.create_bucket
        if type(bucket_name) != str and type(bucket_name) != bytes:

Using type() instead of isinstance() for a typecheck.

            raise B2InvalidBucketName("Bucket name must be alphanumeric or '-")
        if type(configuration) != dict and configuration is not None:

Using type() instead of isinstance() for a typecheck.

            raise B2InvalidBucketConfiguration
        params = {
            'bucketName': bucket_name,
            'bucketType': security,
            #TODO: bucketInfo

TODO and FIXME comments should generally be avoided.

            #TODO: corsRules

TODO and FIXME comments should generally be avoided.

            #TODO: lifeCycleRules

TODO and FIXME comments should generally be avoided.

        }
        response = self.connector.make_request(path=path, method='post', params=params, account_id_required=True)

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

        if response.status_code == 200:
            bucket_json = response.json()
            new_bucket = B2Bucket(connector=self.connector, parent_list=self, **bucket_json)
            self._buckets_by_name[bucket_json['bucketName']] = new_bucket
            self._buckets_by_id[bucket_json['bucketId']] = new_bucket
            return new_bucket
        else:
            raise B2Exception.parse(response)

Final newline missing