from azure.identity.aio import ClientSecretCredential from msgraph import GraphServiceClient from msgraph.generated.drives.item.items.item.create_upload_session.create_upload_session_post_request_body import ( CreateUploadSessionPostRequestBody, ) from msgraph.generated.models.drive_item_uploadable_properties import DriveItemUploadableProperties msgraph_settings = { "tenat_id": "your_tenant_id", "client_id": "your_client_id", "client_secret": "your_client_secret", "shrd_id": "your_shrd_id", } credential = ClientSecretCredential( tenant_id=msgraph_settings.tenant_id, client_id=msgraph_settings.client_id, client_secret=msgraph_settings.client_secret, ) client = GraphServiceClient( credentials=credential, scopes=[ "https://graph.microsoft.com/.default", ], ) async def upload_small_file(destination_path: str, file_content): uploadable_properties = DriveItemUploadableProperties( additional_data={"@microsoft.graph.conflictBehavior": "replace"} ) upload_session_request_body = CreateUploadSessionPostRequestBody(item=uploadable_properties) # can be used for normal drive uploads try: upload_session = ( await client.drives.by_drive_id(msgraph_settings.shrd_id) .items.by_drive_item_id(f"root:/{destination_path}:") .create_upload_session.post(upload_session_request_body) ) if upload_session: print(f"Created upload session for {destination_path}") uploaded_file = ( await client.drives.by_drive_id(msgraph_settings.shrd_id) .items.by_drive_item_id(f"root:/{destination_path}:") .content.put(file_content) ) if uploaded_file: print(f"Uploaded file: {destination_path}") except Exception as e: print(f"Error creating upload session: {e}") raise e