frappe-impl-integrations

Frappe 集成

OAuth、Webhook、支付网关、数据导入/导出和外部 API 调用的分步工作流。

版本：v14/v15/v16

---

决策树：选择哪种集成模式？

WHAT ARE YOU INTEGRATING?
│
├─► External service needs to call YOUR Frappe site?
│   ├─► On document events → Webhook (push to external)
│   ├─► External sends data to you → Whitelisted API endpoint
│   └─► External needs user auth → OAuth 2.0 Provider
│
├─► YOUR Frappe site calls an external service?
│   ├─► Needs user-level OAuth consent → Connected App
│   ├─► Server-to-server with API key → make_request / requests
│   └─► Recurring sync → Scheduler + API calls
│
├─► Bulk data in/out?
│   ├─► Import CSV/XLSX → Data Import DocType
│   ├─► Export data → Report Builder / export-csv / API
│   └─► Programmatic bulk → frappe.get_doc().insert()
│
├─► Payment processing?
│   └─► Payment Request + Payment Gateway controller
│
└─► Real-time vs batch?
    ├─► Real-time → Webhook or API endpoint
    ├─► Near real-time → frappe.enqueue() after event
    └─► Batch → Scheduler task (hourly/daily)

---

工作流 1：OAuth 2.0：Frappe 作为提供方

当外部应用需要“使用 Frappe 登录”或代表用户访问 API 时使用。

第 1 步：配置 OAuth 提供方设置

前往 Setup > Integrations > OAuth Provider Settings：

• Force：始终要求用户确认
• Auto：仅在不存在有效令牌时询问

第 2 步：创建 OAuth 客户端

前往 Setup > Integrations > OAuth Client：

字段	值
App Name	外部应用标识符
Scopes	以空格分隔（例如，openid all）
Redirect URIs	以空格分隔的回调 URL
Default Redirect URI	主要回调 URL
Grant Type	Authorization Code（推荐）或 Implicit
Response Type	Code（用于 Auth Code）或 Token（用于 Implicit）
Skip Authorization	仅对可信的第一方应用勾选

第 3 步：使用生成的端点

端点	URL
Authorize	/api/method/frappe.integrations.oauth2.authorize
Token	/api/method/frappe.integrations.oauth2.get_token
Profile	/api/method/frappe.integrations.oauth2.openid_profile

第 4 步：配置外部应用

# Example: Grafana generic_oauth config
client_id = <generated_client_id>
client_secret = <generated_client_secret>
auth_url = https://your-frappe.com/api/method/frappe.integrations.oauth2.authorize
token_url = https://your-frappe.com/api/method/frappe.integrations.oauth2.get_token
api_url = https://your-frappe.com/api/method/frappe.integrations.oauth2.openid_profile
scopes = openid all

重要规则

• 绝不要为服务端应用使用 Implicit 授权类型——应使用 Authorization Code
• 在生产环境中，始终为所有 OAuth 端点使用 HTTPS
• 绝不要在客户端 JavaScript 中暴露 client_secret

---

工作流 2：连接的应用：Frappe 作为 OAuth 消费方

当你的 Frappe 实例需要代表用户访问外部服务（Google、Microsoft 等）时使用。

第 1 步：创建 Connected App DocType

字段	用途
Name	连接标识符
OpenID Configuration URL	自动获取端点（例如，/.well-known/openid-configuration）
Authorization URI	同意页面 URL（从 OpenID 自动填充）
Token URI	令牌交换 URL（从 OpenID 自动填充）
Redirect URI	自动生成——将其复制到外部提供方
Client ID	来自外部提供方
Client Secret	来自外部提供方
Scopes	所需权限（例如，https://mail.google.com/）

第 2 步：向提供方注册重定向 URI

复制自动生成的 Redirect URI，并在外部提供方的 OAuth 控制台中注册。

第 3 步：添加额外参数（如有需要）

access_type=offline    # Google: enables refresh tokens
prompt=consent         # Google: forces re-consent for refresh token

第 4 步：在代码中使用

import frappe

connected_app = frappe.get_doc("Connected App", "My Google App")
# Initiates OAuth flow — user clicks "Connect to..." button
# After consent, tokens are stored automatically

# Making authenticated calls:
session = connected_app.get_oauth2_session()
response = session.get("https://www.googleapis.com/gmail/v1/users/me/messages")

重要规则

• 对 Google API 始终添加 access_type=offline 以获取刷新令牌
• 绝不要手动存储令牌——Connected App 会管理令牌生命周期
• 始终处理 TokenExpiredError——调用 session.refresh_token() 或重新连接

---

工作流 3：Webhook：向外部服务推送通知

第 1 步：创建 Webhook DocType

前往 Integrations > Webhook：

字段	值
DocType	目标文档类型
Doc Event	on_update、after_insert、on_submit、on_cancel、on_trash
Request URL	外部端点
Request Method	POST（默认）
Conditions	可选的 Jinja 过滤条件（例如，doc.status == "Approved"）
Enabled	勾选以启用

第 2 步：配置请求头

添加用于身份验证的自定义请求头：

Authorization: Bearer <api_token>
Content-Type: application/json

第 3 步：配置数据：选择格式

表单 URL 编码：从表格中选择特定字段。

JSON：使用 Jinja 模板构建结构化载荷：

{
  "id": "{{ doc.name }}",
  "total": "{{ doc.grand_total }}",
  "items": {{ doc.items | tojson }},
  "event": "{{ event }}"
}

第 4 步：启用 Webhook 密钥（HMAC 验证）

设置 Webhook Secret——Frappe 会添加 X-Frappe-Webhook-Signature 请求头，其中包含载荷的 base64 编码 HMAC-SHA256 哈希值。

接收方验证（Python 示例）：

import hmac, hashlib, base64

def verify_webhook(payload_body, secret, signature_header):
    expected = base64.b64encode(
        hmac.new(secret.encode(), payload_body, hashlib.sha256).digest()
    ).decode()
    return hmac.compare_digest(expected, signature_header)

关键规则

• 在生产环境中使用 Webhook 时，务必启用 Webhook Secret
• 切勿依赖 webhook 来保证交付；接收端应实现幂等性
• 在 JSON 负载中使用子表数据时，务必使用 | tojson 过滤器
• 每次交付都会创建 Webhook 日志；调试时请查看 Webhook Request Log

---

工作流 4：从 Frappe 调用外部 API

使用 frappe.integrations.utils

from frappe.integrations.utils import make_get_request, make_post_request

# GET request
response = make_get_request(
    "https://api.example.com/data",
    headers={"Authorization": "Bearer token123"}
)

# POST request
response = make_post_request(
    "https://api.example.com/submit",
    data={"key": "value"},
    headers={"Content-Type": "application/json"}
)

直接使用 requests 库

import requests
import frappe

def sync_to_external():
    try:
        response = requests.post(
            "https://api.example.com/endpoint",
            json={"data": "value"},
            timeout=30
        )
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        frappe.log_error(f"API call failed: {e}", "Integration Error")
        raise

关键规则

• 对外部请求务必设置 timeout（建议设为 30s）
• 外部调用务必用 try/except 包装，并使用 frappe.log_error() 记录错误
• 切勿在 validate 或 before_save 中调用外部 API；应使用 on_update + frappe.enqueue()
• 对耗时较长的外部调用，务必使用 frappe.enqueue()，避免阻塞 Web 请求

---

工作流 5：数据导入

通过 UI（Data Import DocType）

1. 前往 Home > Data Import > New
2. 选择 DocType 和导入类型（Insert 或 Update）
3. 下载模板 CSV/XLSX
4. 按照模板格式填写数据
5. 上传并预览
6. 开始导入

CSV 格式规则

ID,Item Name,Item Group,Stock UOM
,Widget A,Products,Nos
,Widget B,Raw Material,Kg

• 第一行：字段标签或 API 字段名
• 对于 Insert，将 ID/name 留空（自动生成）
• 对于 Update，ID 列必须包含现有文档名称
• 子表：重复父行数据，并将子字段作为额外列添加

以编程方式导入

import frappe
from frappe.core.doctype.data_import.data_import import DataImport

# Create Data Import document
di = frappe.get_doc({
    "doctype": "Data Import",
    "reference_doctype": "Item",
    "import_type": "Insert New Records",
    "import_file": "/path/to/file.csv"
})
di.insert()
di.start_import()

关键规则

• 务必下载并使用模板——列顺序和名称必须完全匹配
• 一次切勿导入超过 5,000 行；请分批处理
• 批量导入前，务必先用 5-10 行进行测试
• 导入完成后，务必检查 Import Log 中的逐行错误

---

工作流 6：数据导出

通过 Report Builder

1. 打开任意 DocType 列表视图
2. 应用筛选条件
3. 菜单 > 导出（CSV/Excel）

通过 CLI

bench --site mysite export-csv "Sales Invoice"
bench --site mysite export-doc "Sales Invoice" "INV-001"
bench --site mysite export-json "Sales Invoice" "INV-001"
bench --site mysite export-fixtures --app myapp

以编程方式导出

import frappe

# Export filtered data
data = frappe.get_all("Sales Invoice",
    filters={"status": "Paid", "posting_date": [">", "2024-01-01"]},
    fields=["name", "customer", "grand_total", "posting_date"],
    order_by="posting_date desc",
    limit_page_length=0  # No limit
)

# Convert to CSV
import csv, io
output = io.StringIO()
writer = csv.DictWriter(output, fieldnames=["name", "customer", "grand_total", "posting_date"])
writer.writeheader()
writer.writerows(data)
csv_content = output.getvalue()

---

工作流 7：Frappe REST API 身份验证

API Key + Secret（服务器到服务器）

# Generate via User > API Access > Generate Keys
curl -H "Authorization: token api_key:api_secret" \
  https://your-site.com/api/resource/Sales%20Invoice

OAuth Bearer Token

curl -H "Authorization: Bearer access_token" \
  https://your-site.com/api/resource/Sales%20Invoice

基于会话（登录）

# Login first
curl -X POST https://your-site.com/api/method/login \
  -d "[email protected]&pwd=password"
# Subsequent requests use session cookie

---

集成模式：同步与异步

模式	使用场景	实现方式
同步	需要立即获得响应	在控制器中直接调用 API
异步（入队）	外部调用 > 5s	frappe.enqueue("myapp.api.sync_record", doc_name=doc.name)
Webhook	事件触发推送	Webhook DocType 配置
定时同步	定期批处理	hooks.py 中的 scheduler_events
实时	实时更新	Socket.IO + frappe.publish_realtime()

重试模式

import frappe
from frappe.utils.background_jobs import get_jobs

def sync_with_retry(doc_name, retry_count=0, max_retries=3):
    try:
        result = call_external_api(doc_name)
        frappe.db.set_value("Sales Invoice", doc_name, "sync_status", "Success")
        frappe.db.commit()
    except Exception as e:
        if retry_count < max_retries:
            frappe.enqueue(
                "myapp.integrations.sync_with_retry",
                doc_name=doc_name,
                retry_count=retry_count + 1,
                queue="short",
                enqueue_after_commit=True
            )
        else:
            frappe.log_error(f"Sync failed after {max_retries} retries: {e}")
            frappe.db.set_value("Sales Invoice", doc_name, "sync_status", "Failed")
            frappe.db.commit()

---

版本差异

功能	V14	V15	V16
Webhook DocType	是	是	是
Connected App	是	是	是
OAuth 2.0 Provider	是	是	是
Data Import（新 UI）	是	是	是
Print Designer	否	是	是
make_get_request	是	是	是
Webhook HMAC	是	是	是

---

参考文件

文件	内容
workflows.md	完整的集成工作流模式
examples.md	所有集成类型的可运行代码示例
anti-patterns.md	常见集成错误及修正方法
decision-tree.md	用于选择集成方式的扩展决策树