Source code for boxsdk.object.upload_session

# coding: utf-8
from __future__ import unicode_literals, absolute_import

import base64
import hashlib
import json

from .base_object import BaseObject
from ..pagination.limit_offset_based_dict_collection import LimitOffsetBasedDictCollection

[docs]class UploadSession(BaseObject): _item_type = 'upload_session' _parent_item_type = 'file'
[docs] def get_url(self, *args): """ Base class override. Endpoint is a little different - it's /files/upload_sessions. :rtype: `unicode` """ return self._session.get_url( '{0}s/{1}s'.format(self._parent_item_type, self._item_type), self._object_id, *args ).replace(self.session.api_config.BASE_API_URL, self.session.api_config.UPLOAD_URL)
[docs] def get_parts(self, limit=None, offset=None): """ Get a list of parts uploaded so far. :param limit: The maximum number of items to return per page. If not specified, then will use the server-side default. :type limit: `int` or None :param offset: The index at which to start returning items. :type offset: `int` or None :returns: Returns a :class:`BoxObjectCollection` object containing the uploaded parts. :rtype: :class:`BoxObjectCollection` """ return LimitOffsetBasedDictCollection( session=self.session, url=self.get_url('parts'), limit=limit, offset=offset, fields=None, return_full_pages=False, )
[docs] def upload_part_bytes(self, part_bytes, offset, total_size, part_content_sha1=None): """ Upload a part of a file. :param part_bytes: Part bytes :type part_bytes: `bytes` :param offset: Offset, in number of bytes, of the part compared to the beginning of the file. This number should be a multiple of the part size. :type offset: `int` :param total_size: The size of the file that this part belongs to. :type total_size: `int` :param part_content_sha1: SHA-1 hash of the part's content. If not specified, this will be calculated. :type part_content_sha1: `bytes` or None :returns: The uploaded part record. :rtype: `dict` """ if part_content_sha1 is None: sha1 = hashlib.sha1() sha1.update(part_bytes) part_content_sha1 = sha1.digest() range_end = min(offset + self.part_size - 1, total_size - 1) # pylint:disable=no-member headers = { 'Content-Type': 'application/octet-stream', 'Digest': 'SHA={0}'.format(base64.b64encode(part_content_sha1).decode('utf-8')), 'Content-Range': 'bytes {0}-{1}/{2}'.format(offset, range_end, total_size), } response = self._session.put( self.get_url(), headers=headers, data=part_bytes, ) return response.json()['part']
[docs] def commit(self, content_sha1, parts=None, file_attributes=None, etag=None): """ Commit a multiput upload. :param content_sha1: SHA-1 hash of the file contents that was uploaded. :type content_sha1: `bytes` :param parts: List of parts that were uploaded. :type parts: `Iterable` of `dict` or None :param file_attributes: A `dict` of attributes to set on the uploaded file. :type file_attributes: `dict` :param etag: If specified, instruct the Box API to delete the folder only if the current version's etag matches. :type etag: `unicode` or None :returns: The newly-uploaded file object. :rtype: :class:`File` """ body = {} if file_attributes is not None: body['attributes'] = file_attributes if parts is not None: body['parts'] = parts else: body['parts'] = [part for part in self.get_parts()] headers = { 'Content-Type': 'application/json', 'Digest': 'SHA={0}'.format(base64.b64encode(content_sha1).decode('utf-8')), } if etag is not None: headers['If-Match'] = etag response = self.get_url('commit'), headers=headers, data=json.dumps(body), ).json() entry = response['entries'][0] return self.translator.translate( session=self._session, response_object=entry, )
[docs] def abort(self): """ Abort an upload session, cancelling the upload and removing any parts that have already been uploaded. :returns: A boolean indication success of the upload abort. :rtype: `bool` """ return self.delete()