如何通过Airflow连接Microsoft Dynamics 365 CRM并获取数据？

可行实现方式 #

Airflow暂无官方集成的Dynamics 365插件，但可以通过以下两种核心方式完成数据获取：

1. 借助MS Graph API + PythonOperator #

Dynamics 365的数据可通过Microsoft Graph API访问，你可以在Airflow中用PythonOperator编写自定义Python代码实现调用：

• 步骤1：获取Azure认证凭据
  在Azure Active Directory中注册应用，获取client_id、client_secret、tenant_id，并为应用授予Dynamics 365的对应权限（如CRM.ReadWrite.All）。
• 步骤2：编写数据获取代码
  使用requests库先获取访问令牌，再调用Dynamics 365的实体接口，示例代码片段：

  import requests
  
  def fetch_dynamics_data():
      # 获取访问令牌
      token_url = "https://login.microsoftonline.com/{你的tenant_id}/oauth2/v2.0/token"
      token_payload = {
          "grant_type": "client_credentials",
          "client_id": "你的client_id",
          "client_secret": "你的client_secret",
          "scope": "https://graph.microsoft.com/.default"
      }
      token_response = requests.post(token_url, data=token_payload)
      access_token = token_response.json()["access_token"]
  
      # 调用Dynamics 365实体API（以账户实体为例）
      dynamics_url = "https://你的组织域名.crm.dynamics.com/api/data/v9.2/accounts"
      headers = {"Authorization": f"Bearer {access_token}"}
      data_response = requests.get(dynamics_url, headers=headers)
      # 按需处理返回数据，比如写入数据库、存储至对象存储
      print(data_response.json())
  

• 步骤3：在Airflow DAG中配置任务
  将上述函数通过PythonOperator加入DAG：

  from airflow import DAG
  from airflow.operators.python import PythonOperator
  from datetime import datetime
  
  default_args = {
      'owner': 'airflow',
      'start_date': datetime(2024, 1, 1)
  }
  
  with DAG('dynamics_365_data_sync', default_args=default_args, schedule_interval='@daily') as dag:
      fetch_task = PythonOperator(
          task_id='fetch_dynamics_data',
          python_callable=fetch_dynamics_data
      )
  

2. 封装自定义Airflow Hook #

可以基于requests或msal库封装自定义Hook，将认证、API调用逻辑复用，提升代码可维护性：

• 继承Airflow的BaseHook，在Airflow Connections中存储Azure AD凭据；
• 在Hook中实现获取令牌、调用Dynamics 365 API的方法，后续在Operator中直接调用Hook即可。

关键注意事项 #

• 确保Airflow所在环境能访问Azure和Dynamics 365的网络地址，必要时开放防火墙端口；
• 权限配置遵循最小权限原则，避免过度授权；
• 敏感凭据（如client_secret）需存储在Airflow Connections或Variables中，禁止硬编码，可通过Variable.get("dynamics_client_secret")获取。

内容的提问来源于stack exchange，提问作者Venkata Lakshmi Narsingarao Ma