backblaze__b2_command_line_tool-488

{
  "status": "ok",
  "stdout": "/workspace/b2/api.py:206:    def get_bucket_by_name(self, bucket_name):\n/workspace/b2/api.py:263:                bucket_id = self.get_bucket_by_name(bucket_name).id_\n/workspace/b2/raw_simulator.py:709:            bucket = self._get_bucket_by_name(bucket_name)\n/workspace/b2/raw_simulator.py:924:    def _get_bucket_by_name(self, bucket_name):\n/workspace/b2/sync/folder.py:230:        self.bucket = api.get_bucket_by_name(bucket_name)\n/workspace/b2/console_tool.py:299:        bucket = self.api.get_bucket_by_name(args.bucketName)\n/workspace/b2/console_tool.py:405:            bucket_id_or_none = self.api.get_bucket_by_name(args.bucket).id_\n/workspace/b2/console_tool.py:433:        bucket = self.api.get_bucket_by_name(args.bucketName)\n/workspace/b2/console_tool.py:529:        bucket = self.api.get_bucket_by_name(args.bucketName)\n/workspace/b2/console_tool.py:656:        bucket = self.api.get_bucket_by_name(args.bucketName)\n/workspace/b2/console_tool.py:691:        bucket = self.api.get_bucket_by_name(args.bucketName)\n/workspace/b2/console_tool.py:733:        bucket = self.api.get_bucket_by_name(args.bucketName)\n/workspace/b2/console_tool.py:779:        bucket = self.api.get_bucket_by_name(args.bucketName)\n/workspace/b2/console_tool.py:802:        bucket = self.api.get_bucket_by_name(args.bucketName)\n/workspace/b2/console_tool.py:932:        bucket = self.api.get_bucket_by_name(args.bucketName)\n/workspace/b2/console_tool.py:983:        bucket = self.api.get_bucket_by_name(args.bucketName)\n/workspace/b2/console_tool.py:1236:        bucket = self.api.get_bucket_by_name(args.bucketName)\n/workspace/b2/console_tool.py:1303:        bucket = self.api.get_bucket_by_name(args.bucketName)\n/workspace/test/test_console_tool.py:388:        bucket = self.b2_api.get_bucket_by_name('my-bucket')\n/workspace/test/test_console_tool.py:395:        bucket = self.b2_api.get_bucket_by_name('my-bucket')\n/workspace/test/test_console_tool.py:603:        bucket = self.b2_api.get_bucket_by_name('my-bucket')\n/workspace/test/test_console_tool.py:610:        bucket = self.b2_api.get_bucket_by_name('my-bucket')\n/workspace/test/test_console_tool.py:794:        bucket = self.b2_api.get_bucket_by_name('my-bucket')\n/workspace/test/test_console_tool.py:829:        bucket = self.b2_api.get_bucket_by_name('my-bucket')\n/workspace/test/test_console_tool.py:873:        bucket = self.b2_api.get_bucket_by_name('my-bucket')\n/workspace/test/test_console_tool.py:913:        bucket = self.b2_api.get_bucket_by_name('my-bucket')\n/workspace/test/test_console_tool.py:1070:        bucket = self.b2_api.get_bucket_by_name('my-bucket')\n/workspace/test/test_policy.py:69:        api.get_bucket_by_name.return_value = bucket\n/workspace/test/test_sync.py:248:        self.api.get_bucket_by_name.return_value = self.bucket\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "b2/api.py",
  "content": "        )\n        return self.transferer.download_file_from_url(url, download_dest, progress_listener, range_)\n\n    def get_bucket_by_id(self, bucket_id):\n        return Bucket(self, bucket_id)\n\n    def get_bucket_by_name(self, bucket_name):\n        \"\"\"\n        Returns the bucket_id for the given bucket_name.\n\n        If we don't already know it from the cache, try fetching it from\n        the B2 service.\n        \"\"\"\n        # If we can get it from the stored info, do that.\n        self.check_bucket_restrictions(bucket_name)\n        id_ = self.cache.get_bucket_id_or_none_from_bucket_name(bucket_name)\n        if id_ is not None:\n            return Bucket(self, id_, name=bucket_name)\n\n        for bucket in self.list_buckets():\n            if bucket.name == bucket_name:\n                return bucket\n        raise NonExistentBucket(bucket_name)\n\n    def delete_bucket(self, bucket):\n        \"\"\"\n        Deletes the bucket remotely.\n        For legacy reasons it returns whatever server sends in response,\n        but API user should not rely on the response: if it doesn't raise\n        an exception, it means that the operation was a success\n        \"\"\"\n        account_id = self.account_info.get_account_id()\n        return self.session.delete_bucket(account_id, bucket.id_)\n\n    def list_buckets(self, bucket_name=None):\n        \"\"\"\n        Calls b2_list_buckets and returns a list of buckets.\n\n        When no bucket name is specified, returns *all* of the buckets\n        in the account.  When a bucket name is given, returns just that\n        bucket.  When authorized with an application key restricted to\n        one bucket, you must specify the bucket name, or the request\n        will be unauthorized.\n\n        :param bucket_name: Optional: the name of the one bucket to return.\n        :return: A list of Bucket objects.\n        \"\"\"\n        account_id = self.account_info.get_account_id()\n        self.check_bucket_restrictions(bucket_name)\n\n        # TEMPORARY work around until we fix the API endpoint bug that things requests\n        # with a bucket name are not authorized.  When it's fixed, well just pass the\n        # bucket name (or None) to the raw API.\n        if bucket_name is None:\n            bucket_id = None\n        else:\n            allowed = self.account_info.get_allowed()\n            if allowed['bucketId'] is not None:\n                # We just checked that if there is a bucket restriction we have a bucket name\n                # and it matches.  So if there's a restriction we know that's the bucket we're\n                # looking for.\n                bucket_id = allowed['bucketId']\n            else:\n                bucket_id = self.get_bucket_by_name(bucket_name).id_\n\n        response = self.session.list_buckets(account_id, bucket_id=bucket_id)\n        buckets = BucketFactory.from_api_response(self, response)\n\n        if bucket_name is not None:\n            # If a bucket_name is specified we don't clear the cache because the other buckets could still\n            # be valid. So we save the one bucket returned from the list_buckets call.\n            for bucket in buckets:\n                self.cache.save_bucket(bucket)\n        else:\n            # Otherwise we want to clear the cache and save the buckets returned from list_buckets\n            # since we just got a new list of all the buckets for this account.\n            self.cache.set_bucket_name_cache(buckets)\n        return buckets\n\n    def list_parts(self, file_id, start_part_number=None, batch_size=None):\n        \"\"\"\n        Generator that yields a Part for each of the parts that have been uploaded.\n\n        :param file_id: the ID of the large file that is not finished\n        :param start_part_number: the first part number to return.  defaults to the first part.\n        :param batch_size: the number of parts to fetch at a time from the server\n        \"\"\"\n        batch_size = batch_size or 100\n        while True:\n            response = self.session.list_parts(file_id, start_part_number, batch_size)\n            for part_dict in response['parts']:\n                yield PartFactory.from_list_parts_dict(part_dict)\n            start_part_number = response.get('nextPartNumber')\n            if start_part_number is None:\n                break\n\n    # delete/cancel\n    def cancel_large_file(self, file_id):\n        response = self.session.cancel_large_file(file_id)\n        return FileVersionInfoFactory.from_cancel_large_file_response(response)\n\n    def delete_file_version(self, file_id, file_name):\n        # filename argument is not first, because one day it may become optional\n        response = self.session.delete_file_version(file_id, file_name)\n        assert response['fileId'] == file_id\n        assert response['fileName'] == file_name\n        return FileIdAndName(file_id, file_name)\n\n    # download\n    def get_download_url_for_fileid(self, file_id):\n        url = url_for_api(self.account_info, 'b2_download_file_by_id')\n        return '%s?fileId=%s' % (url, file_id)\n\n    def get_download_url_for_file_name(self, bucket_name, file_name):\n        \"\"\"\n        Returns a URL to download the given file by name.\n        \"\"\"\n        self.check_bucket_restrictions(bucket_name)\n        return '%s/file/%s/%s' % (\n            self.account_info.get_download_url(), bucket_name, b2_url_encode(file_name)\n        )\n\n    # keys\n    def create_key(\n        self, capabilities, key_name, valid_duration_seconds=None, bucket_id=None, name_prefix=None\n    ):\n        account_id = self.account_info.get_account_id()\n\n        response = self.session.create_key(\n            account_id,\n            capabilities=capabilities,\n            key_name=key_name,\n            valid_duration_seconds=valid_duration_seconds,\n            bucket_id=bucket_id,\n            name_prefix=name_prefix\n        )\n\n        assert set(response['capabilities']) == set(capabilities)\n        assert response['keyName'] == key_name\n\n        return response\n\n    def delete_key(self, application_key_id):\n\n        response = self.session.delete_key(application_key_id=application_key_id)\n        return response\n\n    def list_keys(self, start_application_key_id=None):\n        account_id = self.account_info.get_account_id()\n",
  "start_line": 200,
  "end_line": 349,
  "total_lines": 372,
  "is_truncated": true
}

{
  "status": "ok",
  "filepath": "b2/api.py",
  "content": "######################################################################\n#\n# File: b2/api.py\n#\n# Copyright 2018 Backblaze Inc. All Rights Reserved.\n#\n# License https://www.backblaze.com/using_b2_code.html\n#\n######################################################################\n\nimport six\n\nfrom .account_info.sqlite_account_info import SqliteAccountInfo\nfrom .account_info.exception import MissingAccountData\nfrom .b2http import B2Http\nfrom .bucket import Bucket, BucketFactory\nfrom .cache import AuthInfoCache, DummyCache\nfrom .transferer import Transferer\nfrom .exception import NonExistentBucket, RestrictedBucket\nfrom .file_version import FileVersionInfoFactory, FileIdAndName\nfrom .part import PartFactory\nfrom .raw_api import B2RawApi\nfrom .session import B2Session\nfrom .utils import B2TraceMeta, b2_url_encode, limit_trace_arguments\n\ntry:\n    import concurrent.futures as futures\nexcept ImportError:\n    import futures\n\n\ndef url_for_api(info, api_name):\n    if api_name in ['b2_download_file_by_id']:\n        base = info.get_download_url()\n    else:\n        base = info.get_api_url()\n    return base + '/b2api/v1/' + api_name\n\n\n@six.add_metaclass(B2TraceMeta)\nclass B2Api(object):\n    \"\"\"\n    Provides file-level access to B2 services.\n\n    While B2RawApi provides direct access to the B2 web APIs, this\n    class handles several things that simplify the task of uploading\n    and downloading files:\n      - re-acquires authorization tokens when they expire\n      - retrying uploads when an upload URL is busy\n      - breaking large files into parts\n      - emulating a directory structure (B2 buckets are flat)\n\n    Adds an object-oriented layer on top of the raw API, so that\n    buckets and files returned are Python objects with accessor\n    methods.\n\n    Also,  keeps a cache of information needed to access the service,\n    such as auth tokens and upload URLs.\n    \"\"\"\n\n    def __init__(self, account_info=None, cache=None, raw_api=None, max_upload_workers=10):\n        \"\"\"\n        Initializes the API using the given account info.\n        :param account_info:\n        :param cache:\n        :param raw_api:\n        :return:\n        \"\"\"\n        self.raw_api = raw_api or B2RawApi(B2Http())\n        if account_info is None:\n            account_info = SqliteAccountInfo()\n            if cache is None:\n                cache = AuthInfoCache(account_info)\n        self.session = B2Session(self, self.raw_api)\n        self.transferer = Transferer(self.session, account_info)\n        self.account_info = account_info\n        if cache is None:\n            cache = DummyCache()\n        self.cache = cache\n        self.upload_executor = None\n        self.max_workers = max_upload_workers\n\n    def set_thread_pool_size(self, max_workers):\n        \"\"\"\n        Sets the size of the thread pool to use for uploads and downloads.\n\n        Must be called before any work starts, or the thread pool will get\n        the default size of 1.\n        \"\"\"\n        if self.upload_executor is not None:\n            raise Exception('thread pool already created')\n        self.max_workers = max_workers\n\n    def get_thread_pool(self):\n        \"\"\"\n        Returns the thread pool executor to use for uploads and downloads.\n        \"\"\"\n        if self.upload_executor is None:\n            self.upload_executor = futures.ThreadPoolExecutor(max_workers=self.max_workers)\n        return self.upload_executor\n\n    def authorize_automatically(self):\n        try:\n            self.authorize_account(\n                self.account_info.get_realm(),\n                self.account_info.get_account_id(),\n                self.account_info.get_application_key(),\n            )\n        except MissingAccountData:\n            return False\n        return True\n\n    @limit_trace_arguments(only=('self', 'realm'))\n    def authorize_account(self, realm, account_id_or_key_id, application_key):\n        # Clean up any previous account info if it was for a different account.\n        try:\n            old_account_id = self.account_info.get_account_id()\n            old_realm = self.account_info.get_realm()\n            if account_id_or_key_id != old_account_id or realm != old_realm:\n                self.cache.clear()\n        except MissingAccountData:\n            self.cache.clear()\n\n        # Authorize\n        realm_url = self.account_info.REALM_URLS[realm]\n        response = self.raw_api.authorize_account(realm_url, account_id_or_key_id, application_key)\n        allowed = response['allowed']\n\n        # If there is a bucket restriction, get the name of the bucket.\n        # And, if we have a list of the one allowed bucket, go ahead and\n        # save it.\n        if 'listBuckets' in allowed['capabilities']:\n            if allowed['bucketId'] is not None:\n                allowed_bucket_response = self.raw_api.list_buckets(\n                    api_url=response['apiUrl'],\n                    account_auth_token=response['authorizationToken'],\n                    account_id=response['accountId'],\n                    bucket_id=allowed['bucketId'],\n                )\n                allowed_bucket_list = allowed_bucket_response['buckets']\n                allowed_bucket_count = len(allowed_bucket_list)\n                assert allowed_bucket_count in [0, 1]\n                if allowed_bucket_count == 0:\n                    # bucket has been deleted since the key was made\n                    allowed['bucketName'] = None\n                else:\n                    allowed['bucketName'] = allowed_bucket_list[0]['bucketName']\n                    self.cache.set_bucket_name_cache(\n                        BucketFactory.from_api_response(self, allowed_bucket_response)\n                    )",
  "start_line": 1,
  "end_line": 150,
  "total_lines": 372,
  "is_truncated": true
}

{
  "status": "ok",
  "stdout": "359:    def check_bucket_restrictions(self, bucket_name):\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "b2/api.py",
  "content": "        return self.session.list_keys(\n            account_id, max_key_count=1000, start_application_key_id=start_application_key_id\n        )\n\n    # other\n    def get_file_info(self, file_id):\n        \"\"\" legacy interface which just returns whatever remote API returns \"\"\"\n        return self.session.get_file_info(file_id)\n\n    def check_bucket_restrictions(self, bucket_name):\n        \"\"\"\n        Checks to see if the allowed field from authorize-account\n        has a bucket restriction.\n\n        If it does, does the bucket_name for a given api call match that.\n        If not it raises a RestrictedBucket error.\n        \"\"\"\n        allowed = self.account_info.get_allowed()\n        allowed_bucket_name = allowed['bucketName']\n\n        if allowed_bucket_name is not None:\n            if allowed_bucket_name != bucket_name:\n                raise RestrictedBucket(allowed_bucket_name)",
  "start_line": 350,
  "end_line": 372,
  "total_lines": 372,
  "is_truncated": false
}

{
  "status": "ok",
  "filepath": "b2/session.py",
  "content": "######################################################################\n#\n# File: b2/session.py\n#\n# Copyright 2018 Backblaze Inc. All Rights Reserved.\n#\n# License https://www.backblaze.com/using_b2_code.html\n#\n######################################################################\n\nimport functools\n\nfrom b2.exception import (InvalidAuthToken, Unauthorized)\nfrom b2.raw_api import ALL_CAPABILITIES\n\n\nclass B2Session(object):\n    \"\"\"\n        A *magic* facade that supplies the correct api_url and account_auth_token\n        to methods of underlying raw_api and reauthorizes if necessary\n    \"\"\"\n\n    def __init__(self, api, raw_api):\n        self._api = api  # for reauthorization\n        self.raw_api = raw_api\n\n    def __getattr__(self, name):\n        f = getattr(self.raw_api, name)\n\n        @functools.wraps(f)\n        def wrapper(*args, **kwargs):\n            auth_failure_encountered = False\n            # download_by_name uses different URLs\n            url_factory = kwargs.pop('url_factory', self._api.account_info.get_api_url)\n            while 1:\n                api_url = url_factory()\n                account_auth_token = self._api.account_info.get_account_auth_token()\n                try:\n                    return f(api_url, account_auth_token, *args, **kwargs)\n                except InvalidAuthToken:\n                    if not auth_failure_encountered:\n                        auth_failure_encountered = True\n                        reauthorization_success = self._api.authorize_automatically()\n                        if reauthorization_success:\n                            continue\n                        # TODO: exception chaining could be added here\n                        #       to help debug reauthorization failures\n                    raise\n                except Unauthorized as e:\n                    raise self._add_app_key_info_to_unauthorized(e)\n\n        return wrapper\n\n    def _add_app_key_info_to_unauthorized(self, unauthorized):\n        \"\"\"\n        Takes an Unauthorized error and adds information from the application key\n        about why it might have failed.\n        \"\"\"\n        # What's allowed?\n        allowed = self._api.account_info.get_allowed()\n        capabilities = allowed['capabilities']\n        bucket_name = allowed['bucketName']\n        name_prefix = allowed['namePrefix']\n\n        # Make a list of messages about the application key restrictions\n        key_messages = []\n        if set(capabilities) != set(ALL_CAPABILITIES):\n            key_messages.append(\"with capabilities '\" + ','.join(capabilities) + \"'\")\n        if bucket_name is not None:\n            key_messages.append(\"restricted to bucket '\" + bucket_name + \"'\")\n        if name_prefix is not None:\n            key_messages.append(\"restricted to files that start with '\" + name_prefix + \"'\")\n        if not key_messages:\n            key_messages.append('with no restrictions')\n\n        # Make a new message\n        new_message = unauthorized.message\n        if new_message == '':\n            new_message = 'unauthorized'\n        new_message += ' for application key ' + ', '.join(key_messages)\n\n        return Unauthorized(new_message, unauthorized.code)",
  "start_line": 1,
  "end_line": 82,
  "total_lines": 82,
  "is_truncated": false
}

{
  "status": "ok",
  "filepath": "b2/raw_api.py",
  "content": "######################################################################\n#\n# File: b2/raw_api.py\n#\n# Copyright 2018 Backblaze Inc. All Rights Reserved.\n#\n# License https://www.backblaze.com/using_b2_code.html\n#\n######################################################################\n\nfrom __future__ import print_function\n\nimport base64\nimport os\nimport random\nimport re\nimport sys\nimport time\nimport traceback\nfrom abc import ABCMeta, abstractmethod\n\nimport six\n\nfrom .b2http import B2Http\nfrom .exception import UnusableFileName\nfrom .utils import b2_url_encode, hex_sha1_of_stream\n\n# All possible capabilities\nALL_CAPABILITIES = [\n    'listKeys',\n    'writeKeys',\n    'deleteKeys',\n    'listBuckets',\n    'writeBuckets',\n    'deleteBuckets',\n    'listFiles',\n    'readFiles',\n    'shareFiles',\n    'writeFiles',\n    'deleteFiles',\n]\n\n# Standard names for file info entries\nSRC_LAST_MODIFIED_MILLIS = 'src_last_modified_millis'\n\n# Special X-Bz-Content-Sha1 value to verify checksum at the end\nHEX_DIGITS_AT_END = 'hex_digits_at_end'\n\n\n@six.add_metaclass(ABCMeta)\nclass AbstractRawApi(object):\n    \"\"\"\n    Direct access to the B2 web apis.\n    \"\"\"\n\n    @abstractmethod\n    def cancel_large_file(self, api_url, account_auth_token, file_id):\n        pass\n\n    @abstractmethod\n    def delete_bucket(self, api_url, account_auth_token, account_id, bucket_id):\n        pass\n\n    @abstractmethod\n    def delete_file_version(self, api_url, account_auth_token, file_id, file_name):\n        pass\n\n    @abstractmethod\n    def finish_large_file(self, api_url, account_auth_token, file_id, part_sha1_array):\n        pass\n\n    @abstractmethod\n    def get_upload_part_url(self, api_url, account_auth_token, file_id):\n        pass\n\n    @abstractmethod\n    def hide_file(self, api_url, account_auth_token, bucket_id, file_name):\n        pass\n\n    @abstractmethod\n    def list_parts(self, api_url, account_auth_token, file_id, start_part_number, max_part_count):\n        pass\n\n    @abstractmethod\n    def list_unfinished_large_files(\n        self, api_url, account_auth_token, bucket_id, start_file_id=None, max_file_count=None\n    ):\n        pass\n\n    @abstractmethod\n    def start_large_file(\n        self, api_url, account_auth_token, bucket_id, file_name, content_type, file_info\n    ):\n        pass\n\n    @abstractmethod\n    def update_bucket(\n        self,\n        api_url,\n        account_auth_token,\n        account_id,\n        bucket_id,\n        bucket_type=None,\n        bucket_info=None,\n        cors_rules=None,\n        lifecycle_rules=None,\n        if_revision_is=None\n    ):\n        pass\n\n    @abstractmethod\n    def upload_part(\n        self, upload_url, upload_auth_token, part_number, content_length, sha1_sum, input_stream\n    ):\n        pass\n\n    def get_download_url_by_id(self, download_url, account_auth_token, file_id):\n        return download_url + '/b2api/v1/b2_download_file_by_id?fileId=' + file_id\n\n    def get_download_url_by_name(self, download_url, account_auth_token, bucket_name, file_name):\n        return download_url + '/file/' + bucket_name + '/' + b2_url_encode(file_name)\n\n\nclass B2RawApi(AbstractRawApi):\n    \"\"\"\n    Provides access to the B2 web APIs, exactly as they are provided by B2.\n\n    Requires that you provide all necessary URLs and auth tokens for each call.\n\n    Each API call decodes the returned JSON and returns a dict.\n\n    For details on what each method does, see the B2 docs:\n        https://www.backblaze.com/b2/docs/\n\n    This class is intended to be a super-simple, very thin layer on top\n    of the HTTP calls.  It can be mocked-out for testing higher layers.\n    And this class can be tested by exercising each call just once,\n    which is relatively quick.\n\n    All public methods of this class except authorize_account shall accept\n    api_url and account_info as first two positional arguments. This is needed\n    for B2Session magic.\n    \"\"\"\n\n    def __init__(self, b2_http):\n        self.b2_http = b2_http\n\n    def _post_json(self, base_url, api_name, auth, **params):\n        \"\"\"\n        Helper method for calling an API with the given auth and params.",
  "start_line": 1,
  "end_line": 150,
  "total_lines": 793,
  "is_truncated": true
}

{
  "status": "ok",
  "stdout": "287:    def list_buckets(\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "b2/raw_api.py",
  "content": "    def list_buckets(\n        self,\n        api_url,\n        account_auth_token,\n        account_id,\n        bucket_id=None,\n        bucket_name=None,\n    ):\n        return self._post_json(\n            api_url,\n            'b2_list_buckets',\n            account_auth_token,\n            accountId=account_id,\n            bucketTypes=['all'],\n            bucketId=bucket_id,\n            bucketName=bucket_name,\n        )\n\n    def list_file_names(\n        self, api_url, account_auth_token, bucket_id, start_file_name=None, max_file_count=None\n    ):\n        return self._post_json(\n            api_url,\n            'b2_list_file_names',",
  "start_line": 287,
  "end_line": 310,
  "total_lines": 793,
  "is_truncated": false
}