Microsoft Graph 文件系统¶

Microsoft Graph 文件系统通过 Airflow 的 ObjectStoragePath 接口提供对 OneDrive、SharePoint 和 Teams 文档库的访问。

支持的 URL 格式

msgraph://connection_id/drive_id/path/to/file
sharepoint://connection_id/drive_id/path/to/file
onedrive://connection_id/drive_id/path/to/file
msgd://connection_id/drive_id/path/to/file

连接配置¶

在 Airflow 中创建 Microsoft Graph 连接，使用以下参数

连接类型: msgraph
主机: Tenant ID
登录: Client ID
密码: Client Secret

连接表单提供额外的配置字段

租户 ID: Azure AD tenant identifier
驱动器 ID: 特定要访问的驱动器（可选 - 留空以进行通用访问）
作用域: OAuth2 作用域（默认：https://graph.microsoft.com/.default)

通过连接额外字段支持的其他 OAuth2 参数

scope: OAuth2 access scope
token_endpoint: Custom token endpoint URL
redirect_uri: OAuth2 redirect URI for authorization code flow
token_endpoint_auth_method: Client authentication method (default: client_secret_basic)
code_challenge_method: PKCE code challenge method (e.g., ‘S256’)
username: Username for password grant flow
password: Password for password grant flow

连接额外字段配置示例

{
    "drive_id": "b!abc123...",
    "scope": "https://graph.microsoft.com/.default",
    "token_endpoint": "https://login.microsoftonline.com/your-tenant/oauth2/v2.0/token",
    "redirect_uri": "https://:8080/callback",
    "token_endpoint_auth_method": "client_secret_post"
}

使用示例¶

读取文件¶

from airflow.sdk.io.path import ObjectStoragePath

# Access a file in OneDrive
path = ObjectStoragePath("onedrive://my_conn/drive123/Documents/data.csv")

# Read file content
with path.open("r") as f:
    content = f.read()

目录操作¶

# List directory contents in SharePoint
sharepoint_path = ObjectStoragePath("sharepoint://sp_conn/site_drive/Shared Documents/")

for item in sharepoint_path.iterdir():
    print(f"Found: {item.name}")
    if item.is_file():
        print(f"  Size: {item.stat().st_size} bytes")

文件操作¶

# Copy file between drives
source = ObjectStoragePath("msgraph://conn1/drive1/source.txt")
target = ObjectStoragePath("msgraph://conn2/drive2/backup/source.txt")
source.copy(target)

# Move file
old_path = ObjectStoragePath("onedrive://conn/drive/temp/file.txt")
new_path = ObjectStoragePath("onedrive://conn/drive/archive/file.txt")
old_path.move(new_path)

# Delete file
file_to_delete = ObjectStoragePath("msgraph://conn/drive/old_data.csv")
file_to_delete.unlink()

写入文件¶

# Write new file
output_path = ObjectStoragePath("sharepoint://sp_conn/docs/reports/report.txt")

with output_path.open("w") as f:
    f.write("Generated report data\n")
    f.write(f"Created at: {datetime.now()}\n")

驱动器发现¶

当您需要为 URL 找到正确的驱动器 ID 时，可以使用 Microsoft Graph API 运算符

from airflow.providers.microsoft.azure.operators.msgraph import MSGraphAsyncOperator

# List all drives for a user
list_drives = MSGraphAsyncOperator(
    task_id="list_drives",
    conn_id="msgraph_conn",
    url="me/drives",
    result_processor=lambda response: [
        {"id": drive["id"], "name": drive["name"]} for drive in response["value"]
    ],
)

URL 方案映射¶

不同的 URL 方案映射到特定的 Microsoft Graph 端点

msgraph:// - 通用 Microsoft Graph 访问
onedrive:// - OneDrive 个人和商业驱动器
sharepoint:// - SharePoint 文档库
msgd:// - msgraph:// 的简写形式

所有方案使用相同的底层 Microsoft Graph API 和身份验证。

Requirements¶

Microsoft Graph 文件系统需要

msgraphfs Python 包
有效的 Azure AD 应用注册并拥有相应权限
针对您租户的 Microsoft Graph API 访问权限

所需的 Microsoft Graph 权限

Files.Read - 用于读取文件
Files.ReadWrite - 用于读取和写入文件
Sites.Read.All - 用于访问 SharePoint 站点（如果使用 sharepoint:// URLs）

跨引用¶

Microsoft Graph API Operators - 用于 API 操作和驱动器发现

参考¶

欲了解更多信息，请参阅