如何用Microsoft Graph Python SDK上传大文件至SharePoint文件夹？

问题描述 #

我需要用msgraph-sdk-python把大文件上传到SharePoint文件夹，但遇到了几个问题：

• Microsoft Graph文档里有PHP版的大文件上传示例，但找不到Python版本；
• 无法获取LargeFileUploadTask类，虽然找到日历模块的上传会话示例，但不知道怎么套用到文件上传上；
• GitHub仓库里有相关issue和临时方案，但没有官方正式解决办法；
• 我知道需要创建UploadSession对象来实现类似PHP示例的功能，但具体操作步骤不清楚；
• 目前用GitHub上的临时方案代码上传，但速度极慢，代码如下：

async def _upload_video(self, video: Video) -> None:
    try:
        if not video.is_downloaded or not video.is_processed:
            raise Exception("Video not downloaded or not processed")

        print(f"Uploading video {video.title}")

        request_body = DriveItem()
        request_body.name = video.title
        request_body.file = File()
        result = await self.app_client.drives.by_drive_id(self.drive_id).items.by_drive_item_id(
            self.trans_video_id).children.post(body=request_body)

        with open(video.path, "rb") as f:
            data = f.read()
        print(f"Uploading data to {video.title}")
        await self.app_client.drives.by_drive_id(self.drive_id).items.by_drive_item_id(result.id).content.put(
            body=data)
        print(f"Video {video.title} has been uploaded")
    except Exception as e:
        print(e)

解决方案 #

你当前的代码是一次性读取整个文件内容后上传，大文件会导致内存占用高、上传速度慢，正确的做法是使用**分片上传会话（UploadSession）**来分块上传大文件。以下是完整的异步实现代码：

核心实现代码 #

from msgraph.generated.models.drive_item import DriveItem
from msgraph.generated.models.create_upload_session_post_request_body import CreateUploadSessionPostRequestBody
import os

async def _upload_large_video(self, video: Video) -> None:
    try:
        if not video.is_downloaded or not video.is_processed:
            raise Exception("Video not downloaded or not processed")

        file_path = video.path
        file_size = os.path.getsize(file_path)
        chunk_size = 32 * 1024 * 1024  # 32MB分块（推荐值，范围5MB-100MB）

        # 1. 创建上传会话
        print(f"创建上传会话：{video.title}")
        request_body = CreateUploadSessionPostRequestBody(
            item=DriveItem(
                name=video.title,
                file=File()
            )
        )
        upload_session = await self.app_client.drives.by_drive_id(self.drive_id).items.by_drive_item_id(
            self.trans_video_id).create_upload_session.post(body=request_body)
        upload_url = upload_session.upload_url

        # 2. 分块上传文件
        print(f"开始分块上传：{video.title}")
        with open(file_path, "rb") as f:
            offset = 0
            while offset < file_size:
                end = min(offset + chunk_size, file_size)
                chunk_data = f.read(chunk_size)
                # 设置分块范围请求头
                headers = {
                    "Content-Length": str(end - offset),
                    "Content-Range": f"bytes {offset}-{end-1}/{file_size}"
                }
                # 发送分块上传请求
                response = await self.app_client.request_adapter.send_async(
                    method="PUT",
                    url=upload_url,
                    headers=headers,
                    content=chunk_data,
                    response_type=DriveItem
                )
                offset = end
                print(f"已完成 {offset}/{file_size} 字节上传")

        print(f"视频 {video.title} 上传完成")
    except Exception as e:
        print(f"上传失败：{e}")

关键说明 #

• 上传会话创建：通过create_upload_session接口获取专属上传URL，该URL可复用至所有分块请求；
• 分块范围控制：每个分块请求必须携带Content-Range头，明确告知Graph当前上传的字节区间；
• 分块大小选择：32MB是官方推荐的分块大小，平衡请求次数与单次请求稳定性；
• 内存优化：分块读取文件，避免一次性加载大文件到内存中，降低资源占用。

这个方案相比一次性上传，不仅能降低内存消耗，还能通过分块实现断点续传（若需要可额外处理），同时大幅提升大文件的上传速度。

内容的提问来源于stack exchange，提问作者M3TALzero1