Upload a small file to a SharePoint site using MS Graph and MSAL

msgraph-upload.py

import requests
import msal
import atexit
import os.path
import urllib.parse

TENANT_ID = '<your tenant id>'
CLIENT_ID = '<your client id>'
SHAREPOINT_HOST_NAME = 'yourcompany.sharepoint.com'
SITE_NAME = '<your site>'

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 file to the 'General' folder -- replace these
    filename = 'Small File.txt'
    folder_path = 'General'

    # check to see if file exists
    path_url = urllib.parse.quote(f'{folder_path}/{filename}')
    result = requests.get(f'{ENDPOINT}/drives/{drive_id}/root:/{path_url}', headers=headers)
    if result.status_code == 200:
        # file exists, replace its contents
        file_info = result.json()
        file_id = file_info['id']
        result = requests.put(
            f'{ENDPOINT}/drives/{drive_id}/items/{file_id}/content',
            headers={
                'Authorization': 'Bearer ' + access_token,
                'Content-type': 'application/binary'
            },
            data=open(filename, 'rb').read()
        )

    elif result.status_code == 404:
        # file does not exist, create a new item
        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.put(
            f'{ENDPOINT}/drives/{drive_id}/items/{folder_id}:/{file_url}:/content',
            headers={
                'Authorization': 'Bearer ' + access_token,
                'Content-type': 'application/binary'
            },
            data=open(filename, 'rb').read()
        )

    result.raise_for_status()

else:
    raise Exception('no access token in result')