Skip to content

Instantly share code, notes, and snippets.

Last active April 15, 2020 11:53
What would you like to do?
Big file upload to Clinked by chunks example
# Example: python3 81b4d2bb-38a9-4ebf-9ce0-ed00949944ff 11 22 /tmp/file.pdf
import os
import sys
import requests
import urllib.parse
import mimetypes
MIN_CHUNK_SIZE = 5_243_000
def main():
# [accesstoken] [account] [group] [file]
if len(sys.argv) < 5:
print('Required arguments: [accesstoken] [account] [group] [file]\n')
access_token = sys.argv[1]
account_id = sys.argv[2]
group_id = sys.argv[3]
file_path = sys.argv[4]
print(f'Uploading {file_path}...\n')
file_name = os.path.basename(file_path)
file_size = os.path.getsize(file_path)
file_extension = os.path.splitext(file_path)[1].lower()
if file_extension not in mimetypes.types_map:
print(f'Failed to guess file type from extension: {file_extension}\n')
file_type = mimetypes.types_map[file_extension]
if file_size < MIN_CHUNK_SIZE:
print(f'File size {file_size} is less than minimal chunk size {MIN_CHUNK_SIZE}\n')
print('Please choose a bigger file for upload\n')
chunk_id = 0
file_position = 0
with open(file_path, mode='rb') as f:
# First chunk will be slightly larger so that all chunks
# are at least 5 MiB
chunk_size = MIN_CHUNK_SIZE + file_size % MIN_CHUNK_SIZE
chunk =
while chunk != b'':
chunk_metadata = {
"fileName": file_name,
"fileType": file_type,
"fileSize": file_size,
"id": chunk_id,
"fileStart": file_position,
"fileEnd": file_position + len(chunk),
print(f'Uploading chunk #{chunk_id}: size={len(chunk)}\n')
r = upload_chunk(access_token, account_id,
group_id, chunk_metadata, chunk)
chunk_id += 1
file_position += len(chunk)
chunk_size = MIN_CHUNK_SIZE
chunk =
file_metadata = {
"name": file_name,
"type": file_type,
"size": file_size,
print('Finishing upload...')
finish_upload(access_token, account_id, group_id, file_metadata)
if r.status_code > 299 or r.status_code < 200:
print('Failed to finish file upload\n')
print(f'Response ({r.status_code}): {r.text}\n')
def upload_chunk(accessToken, accountId, groupId, chunkMetadata, data):
url = f'{BASE_URL}/v2/accounts/{accountId}/groups/{groupId}/files/upload/data'
headers = {
'Authorization': f'Bearer {accessToken}',
'X-File-Name': chunkMetadata['fileName'],
'X-File-Type': chunkMetadata['fileType'],
'X-File-Size': str(chunkMetadata['fileSize']),
'X-File-Chunk': str(chunkMetadata['id']),
'X-File-Start': str(chunkMetadata['fileStart']),
'X-File-End': str(chunkMetadata['fileEnd']),
r =, headers=headers, data=data)
return r
def finish_upload(accessToken, accountId, groupId, fileMetadata):
forwardContext = urllib.parse.quote(
url = f'{BASE_URL}/v2/accounts/{accountId}/groups/{groupId}/files/upload/complete'
url += "?success=true"
url += f"&forwardContext={forwardContext}"
headers = {
'Authorization': f'Bearer {accessToken}',
'X-File-Name': fileMetadata['name'],
'X-File-Type': fileMetadata['type'],
'X-File-Size': str(fileMetadata['size'])
r =, headers=headers, json=fileMetadata)
return r
def validate_chunk_upload_request(request):
if request.status_code > 299 or request.status_code < 200:
print('Failed to upload chunk\n')
print(f'Response ({request.status_code}): {request.text}\n')
# TODO Notify the API about failed upload
if __name__ == "__main__":
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment