frappe-api
Compare original and translation side by side
🇺🇸
Original
English🇨🇳
Translation
ChineseFrappe REST API Development
Frappe REST API开发
Create secure, well-documented REST API endpoints for Frappe Framework v15 following best practices for authentication, permission checking, and input validation.
为Frappe Framework v15创建安全、文档完善的REST API端点,遵循认证、权限检查和输入校验的最佳实践。
When to Use
适用场景
- Building custom REST API endpoints
- Exposing service layer methods via HTTP
- Creating public/private API routes
- Implementing webhook handlers
- Building integrations with external systems
- 构建自定义REST API端点
- 通过HTTP暴露服务层方法
- 创建公开/私有API路由
- 实现Webhook处理器
- 构建与外部系统的集成
Arguments
命令参数
/frappe-api <endpoint_name> [--doctype <doctype>] [--public]Examples:
/frappe-api get_dashboard_stats
/frappe-api create_order --doctype "Sales Order"
/frappe-api webhook_handler --public/frappe-api <endpoint_name> [--doctype <doctype>] [--public]示例:
/frappe-api get_dashboard_stats
/frappe-api create_order --doctype "Sales Order"
/frappe-api webhook_handler --publicProcedure
操作步骤
Step 1: Gather API Requirements
步骤1:收集API需求
Ask the user for:
- Endpoint Name (snake_case, e.g., )
get_dashboard_stats - HTTP Methods supported (GET, POST, PUT, DELETE)
- Authentication Type:
- Token (API Key + Secret)
- Session (Cookie-based)
- OAuth 2.0
- Public (no auth required - use sparingly)
- Parameters - Input parameters with types
- Related DocType (if applicable)
- Allowed Roles (who can access this endpoint)
向用户确认以下信息:
- 端点名称(蛇形命名,例如:)
get_dashboard_stats - 支持的HTTP方法(GET、POST、PUT、DELETE)
- 认证类型:
- Token(API密钥+密钥密码)
- Session(基于Cookie)
- OAuth 2.0
- 公开(无需认证 - 谨慎使用)
- 参数 - 带类型的输入参数
- 关联DocType(如适用)
- 允许的角色(哪些用户可访问该端点)
Step 2: Design API Contract
步骤2:设计API契约
Create the API specification:
yaml
Endpoint: /api/method/<app>.<module>.api.<endpoint_name>
Methods: GET, POST
Auth: Token | Session
Rate Limit: 100 req/min (if applicable)
Parameters:
- name: param1
type: string
required: true
description: Description of param1
- name: param2
type: integer
required: false
default: 10
Response:
200:
description: Success
schema:
message: object
400:
description: Validation Error
403:
description: Permission Denied创建API规范:
yaml
Endpoint: /api/method/<app>.<module>.api.<endpoint_name>
Methods: GET, POST
Auth: Token | Session
Rate Limit: 100 req/min (如适用)
Parameters:
- name: param1
type: string
required: true
description: 参数1的说明
- name: param2
type: integer
required: false
default: 10
Response:
200:
description: 成功
schema:
message: object
400:
description: 校验错误
403:
description: 权限拒绝Step 3: Generate API Module Structure
步骤3:生成API模块结构
Create :
<app>/<module>/api/<endpoint_name>.pypython
"""
<Endpoint Name> API
<Brief description of what this API does>
Endpoints:
GET/POST /api/method/<app>.<module>.api.<endpoint_name>.<method_name>
Authentication:
Token: Authorization: token api_key:api_secret
Session: Cookie-based after login
Example:
curl -X POST "https://site.com/api/method/<app>.<module>.api.<endpoint_name>.create" \
-H "Authorization: token api_key:api_secret" \
-H "Content-Type: application/json" \
-d '{"title": "Test"}'
"""
import frappe
from frappe import _
from frappe.utils import cint, cstr, flt
from typing import Optional, Any
from <app>.<module>.services.<service>_service import <Service>Service创建:
<app>/<module>/api/<endpoint_name>.pypython
"""
<端点名称> API
<该API功能的简要说明>
Endpoints:
GET/POST /api/method/<app>.<module>.api.<endpoint_name>.<method_name>
Authentication:
Token: Authorization: token api_key:api_secret
Session: 登录后基于Cookie
Example:
curl -X POST "https://site.com/api/method/<app>.<module>.api.<endpoint_name>.create" \
-H "Authorization: token api_key:api_secret" \
-H "Content-Type: application/json" \
-d '{"title": "Test"}'
"""
import frappe
from frappe import _
from frappe.utils import cint, cstr, flt
from typing import Optional, Any
from <app>.<module>.services.<service>_service import <Service>Service──────────────────────────────────────────────────────────────────────────────
──────────────────────────────────────────────────────────────────────────────
API Endpoints
API Endpoints
v15 TYPE ANNOTATION VALIDATION:
v15 类型注解校验:
Frappe v15 automatically validates function parameter types based on
Frappe v15会基于Python类型提示自动校验函数参数类型。例如,如果你声明limit: int
,传入非整数会自动触发校验错误。
limit: intPython type hints. For example, if you declare limit: int
, passing
limit: int—
a non-integer will raise a validation error automatically.
—
TRANSACTION HANDLING:
事务处理:
Frappe automatically commits on successful POST/PUT requests and
Frappe会在POST/PUT请求成功时自动提交,异常时自动回滚。很少需要手动调用frappe.db.commit()。
rolls back on exceptions. Manual frappe.db.commit() is rarely needed.
──────────────────────────────────────────────────────────────────────────────
──────────────────────────────────────────────────────────────────────────────
—
@frappe.whitelist()
def get(name: str) -> dict:
"""
Get single document by name.
Args:
name: Document name/ID
Returns:
Document data
Raises:
frappe.DoesNotExistError: Document not found
frappe.PermissionError: No read permission
Example:
GET /api/method/<app>.<module>.api.<endpoint>.get?name=DOC-00001
"""
_check_permission("<DocType>", "read")
service = <Service>Service()
return {
"success": True,
"data": service.get(name)
}@frappe.whitelist()
def get_list(
status: Optional[str] = None,
limit: int = 20,
offset: int = 0
) -> dict:
"""
Get list of documents with optional filtering.
Args:
status: Filter by status
limit: Maximum records to return (default: 20, max: 100)
offset: Skip N records for pagination
Returns:
List of documents with pagination info
Example:
GET /api/method/<app>.<module>.api.<endpoint>.get_list?status=Draft&limit=10
"""
_check_permission("<DocType>", "read")
# Validate and sanitize inputs
limit = min(cint(limit) or 20, 100) # Cap at 100
offset = max(cint(offset), 0)
service = <Service>Service()
filters = {}
if status:
filters["status"] = status
data = service.repo.get_list(
filters=filters,
fields=["name", "title", "status", "date", "modified"],
limit=limit,
offset=offset
)
total = service.repo.get_count(filters)
return {
"success": True,
"data": data,
"pagination": {
"total": total,
"limit": limit,
"offset": offset,
"has_more": (offset + limit) < total
}
}@frappe.whitelist(methods=["POST"])
def create(
title: str,
date: Optional[str] = None,
description: Optional[str] = None
) -> dict:
"""
Create new document.
Args:
title: Document title (required)
date: Date in YYYY-MM-DD format
description: Optional description
Returns:
Created document data
Raises:
frappe.ValidationError: Invalid input data
frappe.PermissionError: No create permission
Example:
POST /api/method/<app>.<module>.api.<endpoint>.create
Body: {"title": "New Document", "date": "2024-01-15"}
"""
_check_permission("<DocType>", "create")
# Validate required fields
if not title or not cstr(title).strip():
frappe.throw(_("Title is required"), frappe.ValidationError)
service = <Service>Service()
result = service.create({
"title": cstr(title).strip(),
"date": date or frappe.utils.today(),
"description": description
})
frappe.db.commit()
return {
"success": True,
"message": _("Document created successfully"),
"data": result
}@frappe.whitelist(methods=["PUT", "POST"])
def update(
name: str,
title: Optional[str] = None,
status: Optional[str] = None,
description: Optional[str] = None
) -> dict:
"""
Update existing document.
Args:
name: Document name (required)
title: New title
status: New status
description: New description
Returns:
Updated document data
Example:
PUT /api/method/<app>.<module>.api.<endpoint>.update
Body: {"name": "DOC-00001", "title": "Updated Title"}
"""
_check_permission("<DocType>", "write")
if not name:
frappe.throw(_("Document name is required"), frappe.ValidationError)
# Build update data from provided fields
update_data = {}
if title is not None:
update_data["title"] = cstr(title).strip()
if status is not None:
update_data["status"] = status
if description is not None:
update_data["description"] = description
if not update_data:
frappe.throw(_("No fields to update"), frappe.ValidationError)
service = <Service>Service()
result = service.update(name, update_data)
frappe.db.commit()
return {
"success": True,
"message": _("Document updated successfully"),
"data": result
}@frappe.whitelist(methods=["DELETE", "POST"])
def delete(name: str) -> dict:
"""
Delete document.
Args:
name: Document name to delete
Returns:
Success confirmation
Example:
DELETE /api/method/<app>.<module>.api.<endpoint>.delete?name=DOC-00001
"""
_check_permission("<DocType>", "delete")
if not name:
frappe.throw(_("Document name is required"), frappe.ValidationError)
service = <Service>Service()
service.repo.delete(name)
frappe.db.commit()
return {
"success": True,
"message": _("Document deleted successfully")
}@frappe.whitelist(methods=["POST"])
def submit(name: str) -> dict:
"""
Submit document for processing.
Args:
name: Document name to submit
Returns:
Submitted document data
Example:
POST /api/method/<app>.<module>.api.<endpoint>.submit
Body: {"name": "DOC-00001"}
"""
_check_permission("<DocType>", "submit")
service = <Service>Service()
result = service.submit(name)
frappe.db.commit()
return {
"success": True,
"message": _("Document submitted successfully"),
"data": result
}@frappe.whitelist(methods=["POST"])
def cancel(name: str, reason: Optional[str] = None) -> dict:
"""
Cancel submitted document.
Args:
name: Document name to cancel
reason: Cancellation reason
Returns:
Cancelled document data
Example:
POST /api/method/<app>.<module>.api.<endpoint>.cancel
Body: {"name": "DOC-00001", "reason": "Customer request"}
"""
_check_permission("<DocType>", "cancel")
service = <Service>Service()
result = service.cancel(name, reason)
frappe.db.commit()
return {
"success": True,
"message": _("Document cancelled successfully"),
"data": result
}@frappe.whitelist()
def get(name: str) -> dict:
"""
根据名称获取单个文档。
Args:
name: 文档名称/ID
Returns:
文档数据
Raises:
frappe.DoesNotExistError: 文档不存在
frappe.PermissionError: 无读取权限
Example:
GET /api/method/<app>.<module>.api.<endpoint>.get?name=DOC-00001
"""
_check_permission("<DocType>", "read")
service = <Service>Service()
return {
"success": True,
"data": service.get(name)
}@frappe.whitelist()
def get_list(
status: Optional[str] = None,
limit: int = 20,
offset: int = 0
) -> dict:
"""
获取文档列表,支持可选过滤。
Args:
status: 按状态过滤
limit: 返回的最大记录数(默认:20,最大值:100)
offset: 分页跳过的记录数
Returns:
带分页信息的文档列表
Example:
GET /api/method/<app>.<module>.api.<endpoint>.get_list?status=Draft&limit=10
"""
_check_permission("<DocType>", "read")
# 校验并清理输入
limit = min(cint(limit) or 20, 100) # 上限为100
offset = max(cint(offset), 0)
service = <Service>Service()
filters = {}
if status:
filters["status"] = status
data = service.repo.get_list(
filters=filters,
fields=["name", "title", "status", "date", "modified"],
limit=limit,
offset=offset
)
total = service.repo.get_count(filters)
return {
"success": True,
"data": data,
"pagination": {
"total": total,
"limit": limit,
"offset": offset,
"has_more": (offset + limit) < total
}
}@frappe.whitelist(methods=["POST"])
def create(
title: str,
date: Optional[str] = None,
description: Optional[str] = None
) -> dict:
"""
创建新文档。
Args:
title: 文档标题(必填)
date: 日期,格式为YYYY-MM-DD
description: 可选说明
Returns:
创建后的文档数据
Raises:
frappe.ValidationError: 输入数据无效
frappe.PermissionError: 无创建权限
Example:
POST /api/method/<app>.<module>.api.<endpoint>.create
Body: {"title": "New Document", "date": "2024-01-15"}
"""
_check_permission("<DocType>", "create")
# 校验必填字段
if not title or not cstr(title).strip():
frappe.throw(_("标题为必填项"), frappe.ValidationError)
service = <Service>Service()
result = service.create({
"title": cstr(title).strip(),
"date": date or frappe.utils.today(),
"description": description
})
frappe.db.commit()
return {
"success": True,
"message": _("文档创建成功"),
"data": result
}@frappe.whitelist(methods=["PUT", "POST"])
def update(
name: str,
title: Optional[str] = None,
status: Optional[str] = None,
description: Optional[str] = None
) -> dict:
"""
更新现有文档。
Args:
name: 文档名称(必填)
title: 新标题
status: 新状态
description: 新说明
Returns:
更新后的文档数据
Example:
PUT /api/method/<app>.<module>.api.<endpoint>.update
Body: {"name": "DOC-00001", "title": "Updated Title"}
"""
_check_permission("<DocType>", "write")
if not name:
frappe.throw(_("文档名称为必填项"), frappe.ValidationError)
# 从提供的字段构建更新数据
update_data = {}
if title is not None:
update_data["title"] = cstr(title).strip()
if status is not None:
update_data["status"] = status
if description is not None:
update_data["description"] = description
if not update_data:
frappe.throw(_("无可更新字段"), frappe.ValidationError)
service = <Service>Service()
result = service.update(name, update_data)
frappe.db.commit()
return {
"success": True,
"message": _("文档更新成功"),
"data": result
}@frappe.whitelist(methods=["DELETE", "POST"])
def delete(name: str) -> dict:
"""
删除文档。
Args:
name: 要删除的文档名称
Returns:
成功确认信息
Example:
DELETE /api/method/<app>.<module>.api.<endpoint>.delete?name=DOC-00001
"""
_check_permission("<DocType>", "delete")
if not name:
frappe.throw(_("文档名称为必填项"), frappe.ValidationError)
service = <Service>Service()
service.repo.delete(name)
frappe.db.commit()
return {
"success": True,
"message": _("文档删除成功")
}@frappe.whitelist(methods=["POST"])
def submit(name: str) -> dict:
"""
提交文档进行处理。
Args:
name: 要提交的文档名称
Returns:
提交后的文档数据
Example:
POST /api/method/<app>.<module>.api.<endpoint>.submit
Body: {"name": "DOC-00001"}
"""
_check_permission("<DocType>", "submit")
service = <Service>Service()
result = service.submit(name)
frappe.db.commit()
return {
"success": True,
"message": _("文档提交成功"),
"data": result
}@frappe.whitelist(methods=["POST"])
def cancel(name: str, reason: Optional[str] = None) -> dict:
"""
取消已提交的文档。
Args:
name: 要取消的文档名称
reason: 取消原因
Returns:
取消后的文档数据
Example:
POST /api/method/<app>.<module>.api.<endpoint>.cancel
Body: {"name": "DOC-00001", "reason": "Customer request"}
"""
_check_permission("<DocType>", "cancel")
service = <Service>Service()
result = service.cancel(name, reason)
frappe.db.commit()
return {
"success": True,
"message": _("文档取消成功"),
"data": result
}──────────────────────────────────────────────────────────────────────────────
──────────────────────────────────────────────────────────────────────────────
Bulk Operations
批量操作
──────────────────────────────────────────────────────────────────────────────
──────────────────────────────────────────────────────────────────────────────
@frappe.whitelist(methods=["POST"])
def bulk_update_status(names: list[str], status: str) -> dict:
"""
Bulk update status for multiple documents.
Args:
names: List of document names
status: New status to set
Returns:
Number of documents updated
Example:
POST /api/method/<app>.<module>.api.<endpoint>.bulk_update_status
Body: {"names": ["DOC-001", "DOC-002"], "status": "Completed"}
"""
_check_permission("<DocType>", "write")
if not names or not isinstance(names, list):
frappe.throw(_("Names must be a non-empty list"), frappe.ValidationError)
valid_statuses = ["Draft", "Pending", "Completed", "Cancelled"]
if status not in valid_statuses:
frappe.throw(
_("Invalid status. Must be one of: {0}").format(", ".join(valid_statuses)),
frappe.ValidationError
)
service = <Service>Service()
count = service.repo.bulk_update_status(names, status)
frappe.db.commit()
return {
"success": True,
"message": _("{0} documents updated").format(count),
"data": {"updated_count": count}
}@frappe.whitelist(methods=["POST"])
def bulk_update_status(names: list[str], status: str) -> dict:
"""
批量更新多个文档的状态。
Args:
names: 文档名称列表
status: 要设置的新状态
Returns:
更新的文档数量
Example:
POST /api/method/<app>.<module>.api.<endpoint>.bulk_update_status
Body: {"names": ["DOC-001", "DOC-002"], "status": "Completed"}
"""
_check_permission("<DocType>", "write")
if not names or not isinstance(names, list):
frappe.throw(_("名称必须为非空列表"), frappe.ValidationError)
valid_statuses = ["Draft", "Pending", "Completed", "Cancelled"]
if status not in valid_statuses:
frappe.throw(
_("无效状态。必须是以下之一:{0}").format(", ".join(valid_statuses)),
frappe.ValidationError
)
service = <Service>Service()
count = service.repo.bulk_update_status(names, status)
frappe.db.commit()
return {
"success": True,
"message": _("{0}个文档已更新").format(count),
"data": {"updated_count": count}
}──────────────────────────────────────────────────────────────────────────────
──────────────────────────────────────────────────────────────────────────────
Helper Functions
辅助函数
──────────────────────────────────────────────────────────────────────────────
──────────────────────────────────────────────────────────────────────────────
def _check_permission(doctype: str, ptype: str, doc: Any = None) -> None:
"""
Check if current user has permission.
Args:
doctype: DocType to check
ptype: Permission type (read, write, create, delete, submit, cancel)
doc: Optional specific document
Raises:
frappe.PermissionError: If permission denied
"""
if not frappe.has_permission(doctype, ptype, doc=doc):
frappe.throw(
_("You don't have permission to {0} {1}").format(ptype, doctype),
frappe.PermissionError
)def _validate_request_data(data: dict, required: list[str]) -> None:
"""
Validate request data has required fields.
Args:
data: Request data dict
required: List of required field names
Raises:
frappe.ValidationError: If required fields missing
"""
missing = [f for f in required if not data.get(f)]
if missing:
frappe.throw(
_("Missing required fields: {0}").format(", ".join(missing)),
frappe.ValidationError
)def _check_permission(doctype: str, ptype: str, doc: Any = None) -> None:
"""
检查当前用户是否有对应权限。
Args:
doctype: 要检查的DocType
ptype: 权限类型(read、write、create、delete、submit、cancel)
doc: 可选的特定文档
Raises:
frappe.PermissionError: 权限被拒绝时抛出
"""
if not frappe.has_permission(doctype, ptype, doc=doc):
frappe.throw(
_("你没有{0} {1}的权限").format(ptype, doctype),
frappe.PermissionError
)def _validate_request_data(data: dict, required: list[str]) -> None:
"""
校验请求数据包含必填字段。
Args:
data: 请求数据字典
required: 必填字段名称列表
Raises:
frappe.ValidationError: 缺少必填字段时抛出
"""
missing = [f for f in required if not data.get(f)]
if missing:
frappe.throw(
_("缺少必填字段:{0}").format(", ".join(missing)),
frappe.ValidationError
)──────────────────────────────────────────────────────────────────────────────
──────────────────────────────────────────────────────────────────────────────
Public Endpoints (No Auth Required - Use with Caution!)
公开端点(无需认证 - 谨慎使用!)
──────────────────────────────────────────────────────────────────────────────
──────────────────────────────────────────────────────────────────────────────
@frappe.whitelist(allow_guest=True)
def ping() -> dict:
"""
Health check endpoint (public).
Returns:
Server status
Example:
GET /api/method/<app>.<module>.api.<endpoint>.ping
"""
return {
"success": True,
"message": "pong",
"timestamp": frappe.utils.now()
}undefined@frappe.whitelist(allow_guest=True)
def ping() -> dict:
"""
健康检查端点(公开)。
Returns:
服务器状态
Example:
GET /api/method/<app>.<module>.api.<endpoint>.ping
"""
return {
"success": True,
"message": "pong",
"timestamp": frappe.utils.now()
}undefinedStep 4: Generate API v2 Endpoints (Optional)
步骤4:生成v2 API端点(可选)
For v2 REST API pattern, create custom routes in :
hooks.pypython
undefined如需v2 REST API模式,在中创建自定义路由:
hooks.pypython
undefinedhooks.py
hooks.py
Override standard DocType REST endpoints
覆盖标准DocType REST端点
override_doctype_dashboards = {
"<DocType>": "<app>.<module>.api.<endpoint>.get_doctype_dashboard"
}
override_doctype_dashboards = {
"<DocType>": "<app>.<module>.api.<endpoint>.get_doctype_dashboard"
}
Custom website routes for cleaner URLs
自定义网站路由以获得更简洁的URL
website_route_rules = [
{"from_route": "/api/v2/<app>/<endpoint>", "to_route": "<app>.<module>.api.<endpoint>.handle_v2"},
]
undefinedwebsite_route_rules = [
{"from_route": "/api/v2/<app>/<endpoint>", "to_route": "<app>.<module>.api.<endpoint>.handle_v2"},
]
undefinedStep 5: Generate API Tests
步骤5:生成API测试
Create :
<app>/<module>/api/test_<endpoint_name>.pypython
"""
Tests for <Endpoint Name> API
"""
import frappe
from frappe.tests import IntegrationTestCase
class TestAPI<EndpointName>(IntegrationTestCase):
"""API integration tests."""
@classmethod
def setUpClass(cls):
super().setUpClass()
cls.test_user = cls._create_test_user()
cls.test_doc = cls._create_test_document()
@classmethod
def _create_test_user(cls):
"""Create test user with API access."""
if frappe.db.exists("User", "test_api@example.com"):
return frappe.get_doc("User", "test_api@example.com")
user = frappe.get_doc({
"doctype": "User",
"email": "test_api@example.com",
"first_name": "Test",
"last_name": "API User",
"send_welcome_email": 0
}).insert(ignore_permissions=True)
user.add_roles("System Manager")
return user
@classmethod
def _create_test_document(cls):
"""Create test document."""
return frappe.get_doc({
"doctype": "<DocType>",
"title": "API Test Document",
"date": frappe.utils.today()
}).insert()
def test_get_returns_document(self):
"""Test GET endpoint returns document."""
from <app>.<module>.api.<endpoint_name> import get
frappe.set_user(self.test_user.name)
result = get(self.test_doc.name)
self.assertTrue(result.get("success"))
self.assertIsNotNone(result.get("data"))
def test_get_list_with_pagination(self):
"""Test GET list with pagination."""
from <app>.<module>.api.<endpoint_name> import get_list
frappe.set_user(self.test_user.name)
result = get_list(limit=5, offset=0)
self.assertTrue(result.get("success"))
self.assertIn("pagination", result)
self.assertLessEqual(len(result["data"]), 5)
def test_create_validates_input(self):
"""Test CREATE validates required fields."""
from <app>.<module>.api.<endpoint_name> import create
frappe.set_user(self.test_user.name)
with self.assertRaises(frappe.ValidationError):
create(title="") # Empty title should fail
def test_create_returns_document(self):
"""Test CREATE returns new document."""
from <app>.<module>.api.<endpoint_name> import create
frappe.set_user(self.test_user.name)
result = create(title="New Test Doc", date=frappe.utils.today())
self.assertTrue(result.get("success"))
self.assertIsNotNone(result["data"].get("name"))
def test_unauthorized_access_denied(self):
"""Test unauthenticated access is denied."""
from <app>.<module>.api.<endpoint_name> import get
frappe.set_user("Guest")
with self.assertRaises(frappe.PermissionError):
get(self.test_doc.name)
def test_ping_public_access(self):
"""Test ping endpoint is publicly accessible."""
from <app>.<module>.api.<endpoint_name> import ping
frappe.set_user("Guest")
result = ping()
self.assertTrue(result.get("success"))
self.assertEqual(result.get("message"), "pong")创建:
<app>/<module>/api/test_<endpoint_name>.pypython
"""
<端点名称> API测试
"""
import frappe
from frappe.tests import IntegrationTestCase
class TestAPI<EndpointName>(IntegrationTestCase):
"""API集成测试。"""
@classmethod
def setUpClass(cls):
super().setUpClass()
cls.test_user = cls._create_test_user()
cls.test_doc = cls._create_test_document()
@classmethod
def _create_test_user(cls):
"""创建具备API访问权限的测试用户。"""
if frappe.db.exists("User", "test_api@example.com"):
return frappe.get_doc("User", "test_api@example.com")
user = frappe.get_doc({
"doctype": "User",
"email": "test_api@example.com",
"first_name": "Test",
"last_name": "API User",
"send_welcome_email": 0
}).insert(ignore_permissions=True)
user.add_roles("System Manager")
return user
@classmethod
def _create_test_document(cls):
"""创建测试文档。"""
return frappe.get_doc({
"doctype": "<DocType>",
"title": "API Test Document",
"date": frappe.utils.today()
}).insert()
def test_get_returns_document(self):
"""测试GET端点返回文档。"""
from <app>.<module>.api.<endpoint_name> import get
frappe.set_user(self.test_user.name)
result = get(self.test_doc.name)
self.assertTrue(result.get("success"))
self.assertIsNotNone(result.get("data"))
def test_get_list_with_pagination(self):
"""测试带分页的GET列表接口。"""
from <app>.<module>.api.<endpoint_name> import get_list
frappe.set_user(self.test_user.name)
result = get_list(limit=5, offset=0)
self.assertTrue(result.get("success"))
self.assertIn("pagination", result)
self.assertLessEqual(len(result["data"]), 5)
def test_create_validates_input(self):
"""测试CREATE接口校验输入。"""
from <app>.<module>.api.<endpoint_name> import create
frappe.set_user(self.test_user.name)
with self.assertRaises(frappe.ValidationError):
create(title="") # 空标题应触发错误
def test_create_returns_document(self):
"""测试CREATE接口返回新文档。"""
from <app>.<module>.api.<endpoint_name> import create
frappe.set_user(self.test_user.name)
result = create(title="New Test Doc", date=frappe.utils.today())
self.assertTrue(result.get("success"))
self.assertIsNotNone(result["data"].get("name"))
def test_unauthorized_access_denied(self):
"""测试未认证访问被拒绝。"""
from <app>.<module>.api.<endpoint_name> import get
frappe.set_user("Guest")
with self.assertRaises(frappe.PermissionError):
get(self.test_doc.name)
def test_ping_public_access(self):
"""测试ping端点可公开访问。"""
from <app>.<module>.api.<endpoint_name> import ping
frappe.set_user("Guest")
result = ping()
self.assertTrue(result.get("success"))
self.assertEqual(result.get("message"), "pong")Step 6: Show API Documentation Preview
步骤6:展示API文档预览
undefinedundefinedAPI Endpoint Preview
API端点预览
Module: <app>.<module>.api.<endpoint_name>
Base URL: /api/method/<app>.<module>.api.<endpoint_name>
模块: <app>.<module>.api.<endpoint_name>
基础URL: /api/method/<app>.<module>.api.<endpoint_name>
Endpoints:
端点列表:
| Method | Endpoint | Auth | Description |
|---|---|---|---|
| GET | .get | Token/Session | Get single document |
| GET | .get_list | Token/Session | List with pagination |
| POST | .create | Token/Session | Create document |
| PUT | .update | Token/Session | Update document |
| DELETE | .delete | Token/Session | Delete document |
| POST | .submit | Token/Session | Submit for processing |
| POST | .cancel | Token/Session | Cancel document |
| POST | .bulk_update_status | Token/Session | Bulk status update |
| GET | .ping | Public | Health check |
| 方法 | 端点 | 认证方式 | 描述 |
|---|---|---|---|
| GET | .get | Token/Session | 获取单个文档 |
| GET | .get_list | Token/Session | 带分页的列表查询 |
| POST | .create | Token/Session | 创建文档 |
| PUT | .update | Token/Session | 更新文档 |
| DELETE | .delete | Token/Session | 删除文档 |
| POST | .submit | Token/Session | 提交文档进行处理 |
| POST | .cancel | Token/Session | 取消文档 |
| POST | .bulk_update_status | Token/Session | 批量更新状态 |
| GET | .ping | 公开 | 健康检查 |
Authentication:
认证方式:
bash
undefinedbash
undefinedToken auth (recommended for integrations)
Token认证(推荐用于集成场景)
curl -H "Authorization: token api_key:api_secret"
https://site.com/api/method/<endpoint>
https://site.com/api/method/<endpoint>
curl -H "Authorization: token api_key:api_secret"
https://site.com/api/method/<endpoint>
https://site.com/api/method/<endpoint>
Session auth (for browser clients)
Session认证(适用于浏览器客户端)
First login, then use session cookie
先登录,然后使用会话Cookie
undefinedundefinedFiles to Create:
待创建文件:
📁 <module>/api/
├── 📄 init.py
├── 📄 <endpoint_name>.py
└── 📄 test_<endpoint_name>.py
Create this API module?
undefined📁 <module>/api/
├── 📄 init.py
├── 📄 <endpoint_name>.py
└── 📄 test_<endpoint_name>.py
是否创建此API模块?
undefinedStep 7: Execute and Verify
步骤7:执行与验证
After approval, create files and run tests:
bash
bench --site <site> run-tests --module "<app>.<module>.api.test_<endpoint_name>"获得用户确认后,创建文件并运行测试:
bash
bench --site <site> run-tests --module "<app>.<module>.api.test_<endpoint_name>"Output Format
输出格式
undefinedundefinedAPI Created
API已创建
Module: <app>.<module>.api.<endpoint_name>
Endpoints: 9
模块: <app>.<module>.api.<endpoint_name>
端点数: 9
Files Created:
已创建文件:
- ✅ <endpoint_name>.py (API endpoints)
- ✅ test_<endpoint_name>.py (API tests)
- ✅ Updated init.py
- ✅ <endpoint_name>.py(API端点)
- ✅ test_<endpoint_name>.py(API测试)
- ✅ 更新__init__.py
cURL Examples:
cURL示例:
bash
undefinedbash
undefinedGet document
获取文档
curl -X GET "https://site.com/api/method/<app>.<module>.api.<endpoint>.get?name=DOC-001"
-H "Authorization: token api_key:api_secret"
-H "Authorization: token api_key:api_secret"
curl -X GET "https://site.com/api/method/<app>.<module>.api.<endpoint>.get?name=DOC-001"
-H "Authorization: token api_key:api_secret"
-H "Authorization: token api_key:api_secret"
Create document
创建文档
curl -X POST "https://site.com/api/method/<app>.<module>.api.<endpoint>.create"
-H "Authorization: token api_key:api_secret"
-H "Content-Type: application/json"
-d '{"title": "New Document"}'
-H "Authorization: token api_key:api_secret"
-H "Content-Type: application/json"
-d '{"title": "New Document"}'
undefinedcurl -X POST "https://site.com/api/method/<app>.<module>.api.<endpoint>.create"
-H "Authorization: token api_key:api_secret"
-H "Content-Type: application/json"
-d '{"title": "New Document"}'
-H "Authorization: token api_key:api_secret"
-H "Content-Type: application/json"
-d '{"title": "New Document"}'
undefinedNext Steps:
后续步骤:
- Create API keys: Setup > User > API Access
- Test endpoints with curl or Postman
- Run API tests:
bench --site <site> run-tests --module <test_module>
undefined- 创建API密钥:设置 > 用户 > API访问
- 使用curl或Postman测试端点
- 运行API测试:
bench --site <site> run-tests --module <test_module>
undefinedRules
规则
- Always Check Permissions — Every endpoint must call first
_check_permission() - Validate All Input — Never trust user input, validate and sanitize everything
- Type Annotations — Use Python type hints for v15 auto-validation
- Transaction Handling — Frappe auto-commits on successful requests; manual rarely needed except in background jobs
frappe.db.commit() - Public Endpoints — Use sparingly, only for truly public data
allow_guest=True - Error Handling — Use with appropriate exception types
frappe.throw() - Documentation — Every endpoint must have docstring with Args/Returns/Example
- ALWAYS Confirm — Never create files without explicit user approval
- 始终检查权限 — 每个端点必须首先调用
_check_permission() - 校验所有输入 — 绝不信任用户输入,对所有输入进行校验和清理
- 类型注解 — 使用Python类型提示以利用v15的自动校验功能
- 事务处理 — Frappe会在请求成功时自动提交;除后台任务外,极少需要手动调用
frappe.db.commit() - 公开端点 — 仅在处理真正公开的数据时才使用,务必谨慎
allow_guest=True - 错误处理 — 使用并传入合适的异常类型
frappe.throw() - 文档完善 — 每个端点必须包含带Args/Returns/Example的文档字符串
- 必须确认 — 未经用户明确确认,绝不创建文件