backblaze__b2_command_line_tool-488

{
  "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": "        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": "error",
  "error_type": "CommandError",
  "error_message": "Traceback (most recent call last):\n  File \"<stdin>\", line 4, in <module>\nImportError: cannot import name 'Session' from 'b2.session' (/workspace/b2/session.py)\n",
  "details": {
    "stdout": "",
    "stderr": "Traceback (most recent call last):\n  File \"<stdin>\", line 4, in <module>\nImportError: cannot import name 'Session' from 'b2.session' (/workspace/b2/session.py)\n",
    "exit_code": 1
  }
}

{
  "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": "error",
  "error_type": "CommandError",
  "error_message": "Traceback (most recent call last):\n  File \"<stdin>\", line 49, in <module>\n  File \"<stdin>\", line 27, in test_restricted_bucket\nImportError: cannot import name 'Cache' from 'b2.cache' (/workspace/b2/cache.py)\n",
  "details": {
    "stdout": "",
    "stderr": "Traceback (most recent call last):\n  File \"<stdin>\", line 49, in <module>\n  File \"<stdin>\", line 27, in test_restricted_bucket\nImportError: cannot import name 'Cache' from 'b2.cache' (/workspace/b2/cache.py)\n",
    "exit_code": 1
  }
}

{
  "status": "ok",
  "filepath": "b2/cache.py",
  "content": "######################################################################\n#\n# File: b2/cache.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 abc import (ABCMeta, abstractmethod)\n\nimport six\n\n\n@six.add_metaclass(ABCMeta)\nclass AbstractCache(object):\n    def clear(self):\n        self.set_bucket_name_cache(tuple())\n\n    @abstractmethod\n    def get_bucket_id_or_none_from_bucket_name(self, name):\n        pass\n\n    @abstractmethod\n    def get_bucket_name_or_none_from_allowed(self):\n        pass\n\n    @abstractmethod\n    def save_bucket(self, bucket):\n        pass\n\n    @abstractmethod\n    def set_bucket_name_cache(self, buckets):\n        pass\n\n    def _name_id_iterator(self, buckets):\n        return ((bucket.name, bucket.id_) for bucket in buckets)\n\n\nclass DummyCache(AbstractCache):\n    \"\"\" Cache that does nothing \"\"\"\n\n    def get_bucket_id_or_none_from_bucket_name(self, name):\n        return None\n\n    def get_bucket_name_or_none_from_allowed(self):\n        return None\n\n    def save_bucket(self, bucket):\n        pass\n\n    def set_bucket_name_cache(self, buckets):\n        pass\n\n\nclass InMemoryCache(AbstractCache):\n    \"\"\" Cache that stores the information in memory \"\"\"\n\n    def __init__(self):\n        self.name_id_map = {}\n        self.bucket_name = ''\n\n    def get_bucket_id_or_none_from_bucket_name(self, name):\n        return self.name_id_map.get(name)\n\n    def get_bucket_name_or_none_from_allowed(self):\n        return self.bucket_name\n\n    def save_bucket(self, bucket):\n        self.name_id_map[bucket.name] = bucket.id_\n\n    def set_bucket_name_cache(self, buckets):\n        self.name_id_map = dict(self._name_id_iterator(buckets))\n\n\nclass AuthInfoCache(AbstractCache):\n    \"\"\" Cache that stores data persistently in StoredAccountInfo \"\"\"\n\n    def __init__(self, info):\n        self.info = info\n\n    def get_bucket_id_or_none_from_bucket_name(self, name):\n        return self.info.get_bucket_id_or_none_from_bucket_name(name)\n\n    def get_bucket_name_or_none_from_allowed(self):\n        return self.info.get_bucket_name_or_none_from_allowed()\n\n    def save_bucket(self, bucket):\n        self.info.save_bucket(bucket)\n\n    def set_bucket_name_cache(self, buckets):\n        self.info.refresh_entire_bucket_name_cache(self._name_id_iterator(buckets))",
  "start_line": 1,
  "end_line": 93,
  "total_lines": 93,
  "is_truncated": false
}

{
  "status": "ok",
  "stdout": "Calling get_bucket_by_name('pc-backup')...\nCaught unexpected exception: MissingAccountData: Missing account data: api_url\n",
  "stderr": "Traceback (most recent call last):\n  File \"<stdin>\", line 35, in test_restricted_bucket\n  File \"/workspace/b2/api.py\", line 215, in get_bucket_by_name\n    id_ = self.cache.get_bucket_id_or_none_from_bucket_name(bucket_name)\n  File \"/workspace/b2/session.py\", line 36, in wrapper\n    api_url = url_factory()\n  File \"/workspace/b2/account_info/in_memory.py\", line 24, in inner\n    raise MissingAccountData(function.__name__[4:])\nb2.account_info.exception.MissingAccountData: Missing account data: api_url\n",
  "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:",
  "start_line": 200,
  "end_line": 220,
  "total_lines": 372,
  "is_truncated": false
}

{
  "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",
  "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",
  "stdout": "Calling get_bucket_by_name('pc-backup')...\nCaught expected RestrictedBucket: Application key is restricted to bucket: pc-backup\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "b2/api.py",
  "content": "            name,\n            bucket_type,\n            bucket_info=bucket_info,\n            cors_rules=cors_rules,\n            lifecycle_rules=lifecycle_rules\n        )\n        bucket = BucketFactory.from_api_bucket_dict(self, response)\n        assert name == bucket.name, 'API created a bucket with different name\\\n                                     than requested: %s != %s' % (name, bucket.name)\n        assert bucket_type == bucket.type_, 'API created a bucket with different type\\\n                                             than requested: %s != %s' % (\n            bucket_type, bucket.type_\n        )\n        self.cache.save_bucket(bucket)\n        return bucket\n\n    def download_file_by_id(self, file_id, download_dest, progress_listener=None, range_=None):\n        url = self.session.get_download_url_by_id(\n            file_id,\n            url_factory=self.account_info.get_download_url,\n        )",
  "start_line": 180,
  "end_line": 200,
  "total_lines": 372,
  "is_truncated": false
}

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

{
  "status": "ok",
  "filepath": "b2/api.py",
  "content": "    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": 355,
  "end_line": 372,
  "total_lines": 372,
  "is_truncated": false
}