Microsoft Graph API 算子¶

先决任务¶

要使用这些算子，您必须执行以下操作

使用 AZURE PORTAL 或 AZURE CLI 创建必要的资源。
通过 pip 安装 API 库。
pip install 'apache-airflow[azure]'
可以在安装 Airflow™ 中找到详细信息
设置连接.

MSGraphAsyncOperator¶

使用 MSGraphAsyncOperator 调用 Microsoft Graph API。

以下是使用此算子获取 SharePoint 网站的示例。

tests/system/providers/microsoft/azure/example_msgraph.py[源代码]

    site_task = MSGraphAsyncOperator(
        task_id="news_site",
        conn_id="msgraph_api",
        url="sites/850v1v.sharepoint.com:/sites/news",
        result_processor=lambda context, response: response["id"].split(",")[1],  # only keep site_id
    )

以下是使用此算子获取 SharePoint 网站页面的示例。

tests/system/providers/microsoft/azure/example_msgraph.py[源代码]

    site_pages_task = MSGraphAsyncOperator(
        task_id="news_pages",
        conn_id="msgraph_api",
        api_version="beta",
        url=("sites/%s/pages" % "{{ ti.xcom_pull(task_ids='news_site') }}"),  # noqa: UP031
    )

以下是使用此算子获取 PowerBI 工作区的示例。

tests/system/providers/microsoft/azure/example_powerbi.py[源代码]

    workspaces_task = MSGraphAsyncOperator(
        task_id="workspaces",
        conn_id="powerbi",
        url="myorg/admin/workspaces/modified",
        result_processor=lambda context, response: list(map(lambda workspace: workspace["id"], response)),
    )

以下是使用此算子获取 PowerBI 工作区信息的示例。

tests/system/providers/microsoft/azure/example_powerbi.py[源代码]

    workspaces_info_task = MSGraphAsyncOperator(
        task_id="get_workspace_info",
        conn_id="powerbi",
        url="myorg/admin/workspaces/getInfo",
        method="POST",
        query_parameters={
            "lineage": True,
            "datasourceDetails": True,
            "datasetSchema": True,
            "datasetExpressions": True,
            "getArtifactUsers": True,
        },
        data={"workspaces": workspaces_task.output},
        result_processor=lambda context, response: {"scanId": response["id"]},
    )

以下是使用此运算符刷新 PowerBI 数据集的示例。

tests/system/providers/microsoft/azure/example_powerbi.py[源代码]

    refresh_dataset_task = MSGraphAsyncOperator(
        task_id="refresh_dataset",
        conn_id="powerbi_api",
        url="myorg/groups/{workspaceId}/datasets/{datasetId}/refreshes",
        method="POST",
        path_parameters={
            "workspaceId": "9a7e14c6-9a7d-4b4c-b0f2-799a85e60a51",
            "datasetId": "ffb6096e-d409-4826-aaeb-b5d4b165dc4d",
        },
        data={"type": "full"},  # Needed for enhanced refresh
        result_processor=lambda context, response: response["requestid"],
    )

    refresh_dataset_history_task = MSGraphSensor(
        task_id="refresh_dataset_history",
        conn_id="powerbi_api",
        url="myorg/groups/{workspaceId}/datasets/{datasetId}/refreshes/{refreshId}",
        path_parameters={
            "workspaceId": "9a7e14c6-9a7d-4b4c-b0f2-799a85e60a51",
            "datasetId": "ffb6096e-d409-4826-aaeb-b5d4b165dc4d",
            "refreshId": refresh_dataset_task.output,
        },
        timeout=350.0,
        event_processor=lambda context, event: event["status"] == "Completed",
    )

参考¶

有关更多信息，请查看