Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Upload a large file using an upload session to a SharePoint site using MS Graph and MSAL
import requests
import msal
import atexit
import os.path
import urllib.parse
import os
TENANT_ID = '<your tenant id>'
CLIENT_ID = '<your application id>'
SHAREPOINT_HOST_NAME = 'yourcompany.sharepoint.com'
SITE_NAME = '<your site name>'
AUTHORITY = 'https://login.microsoftonline.com/' + TENANT_ID
ENDPOINT = 'https://graph.microsoft.com/v1.0'
SCOPES = [
'Files.ReadWrite.All',
'Sites.ReadWrite.All',
'User.Read',
'User.ReadBasic.All'
]
cache = msal.SerializableTokenCache()
if os.path.exists('token_cache.bin'):
cache.deserialize(open('token_cache.bin', 'r').read())
atexit.register(lambda: open('token_cache.bin', 'w').write(cache.serialize()) if cache.has_state_changed else None)
app = msal.PublicClientApplication(CLIENT_ID, authority=AUTHORITY, token_cache=cache)
accounts = app.get_accounts()
result = None
if len(accounts) > 0:
result = app.acquire_token_silent(SCOPES, account=accounts[0])
if result is None:
flow = app.initiate_device_flow(scopes=SCOPES)
if 'user_code' not in flow:
raise Exception('Failed to create device flow')
print(flow['message'])
result = app.acquire_token_by_device_flow(flow)
if 'access_token' in result:
access_token = result['access_token']
headers={'Authorization': 'Bearer ' + access_token}
# get the site id
result = requests.get(f'{ENDPOINT}/sites/{SHAREPOINT_HOST_NAME}:/sites/{SITE_NAME}', headers=headers)
result.raise_for_status()
site_info = result.json()
site_id = site_info['id']
# get the drive id
result = requests.get(f'{ENDPOINT}/sites/{site_id}/drive', headers=headers)
result.raise_for_status()
drive_info = result.json()
drive_id = drive_info['id']
# upload a large file to the 'General' folder -- replace these
filename = 'Large File.pdf'
folder_path = 'General'
folder_url = urllib.parse.quote(folder_path)
result = requests.get(f'{ENDPOINT}/drives/{drive_id}/root:/{folder_url}', headers=headers)
result.raise_for_status()
folder_info = result.json()
folder_id = folder_info['id']
file_url = urllib.parse.quote(filename)
result = requests.post(
f'{ENDPOINT}/drives/{drive_id}/items/{folder_id}:/{file_url}:/createUploadSession',
headers=headers,
json={
'@microsoft.graph.conflictBehavior': 'replace',
'description': 'A large test file',
'fileSystemInfo': {'@odata.type': 'microsoft.graph.fileSystemInfo'},
'name': filename
}
)
result.raise_for_status()
upload_session = result.json()
upload_url = upload_session['uploadUrl']
st = os.stat(filename)
size = st.st_size
CHUNK_SIZE = 10485760
chunks = int(size / CHUNK_SIZE) + 1 if size % CHUNK_SIZE > 0 else 0
with open(filename, 'rb') as fd:
start = 0
for chunk_num in range(chunks):
chunk = fd.read(CHUNK_SIZE)
bytes_read = len(chunk)
upload_range = f'bytes {start}-{start + bytes_read - 1}/{size}'
print(f'chunk: {chunk_num} bytes read: {bytes_read} upload range: {upload_range}')
result = requests.put(
upload_url,
headers={
'Content-Length': str(bytes_read),
'Content-Range': upload_range
},
data=chunk
)
result.raise_for_status()
start += bytes_read
else:
raise Exception('no access token in result')
@hadi-muhammad

This comment has been minimized.

Copy link

@hadi-muhammad hadi-muhammad commented Oct 25, 2020

Thank you Keath for sharing your code. I was struggle with authentication of daemon app that uses delegated permissions. Much appreciated!

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.