Big file upload to Clinked by chunks example

clinked_api_upload.py

#!/usr/bin/python3
# Example: python3 clinked_api_upload.py 81b4d2bb-38a9-4ebf-9ce0-ed00949944ff 11 22 /tmp/file.pdf
import os
import sys
import requests
import urllib.parse
import mimetypes

BASE_URL = 'https://api-p1.clinked.com'
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')
        exit()
    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')
        exit()
    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')
        exit()

    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 = f.read(chunk_size)
        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)
            validate_chunk_upload_request(r)
            chunk_id += 1
            file_position += len(chunk)
            chunk_size = MIN_CHUNK_SIZE
            chunk = f.read(chunk_size)

    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')
        print('Exiting...\n')
        exit()

    print('Success!')


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 = requests.post(url=url, headers=headers, data=data)
    return r


def finish_upload(accessToken, accountId, groupId, fileMetadata):
    forwardContext = urllib.parse.quote(
        f"/v2/accounts/{accountId}/groups/{groupId}/files")
    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 = requests.post(url=url, 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
        print('Exiting...\n')
        exit()


if __name__ == "__main__":
    main()