storage.py 文件源码

def get_buckets(self, project_id):
        """Gets all GCS buckets for a project.

        Args:
            project_id (int): The project id for a GCP project.

        Returns:
            list: a list of bucket resource dicts.
            https://cloud.google.com/storage/docs/json_api/v1/buckets

        Raises:
            ApiExecutionError: ApiExecutionError is raised if the call to the
                GCP ClodSQL API fails
        """
        try:
            paged_results = self.repository.buckets.list(project_id,
                                                         projection='full')
            return api_helpers.flatten_list_results(paged_results, 'items')
        except (errors.HttpError, HttpLib2Error) as e:
            LOGGER.warn(api_errors.ApiExecutionError(project_id, e))
            raise api_errors.ApiExecutionError('buckets', e)