Big file upload to Clinked by chunks example
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/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() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment