Loading...
Loading...
Compare original and translation side by side
/frappe-api <endpoint_name> [--doctype <doctype>] [--public]/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 --publicget_dashboard_statsget_dashboard_statsEndpoint: /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 DeniedEndpoint: /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: 权限拒绝<app>/<module>/api/<endpoint_name>.py"""
<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>.py"""
<端点名称> 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>Servicelimit: intlimit: intArgs:
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)
}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
}
}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
}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
}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")
}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
}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
}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)
}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
}
}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
}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
}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": _("文档删除成功")
}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
}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
}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}
}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}
}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
)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
)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
)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
)Returns:
Server status
Example:
GET /api/method/<app>.<module>.api.<endpoint>.ping
"""
return {
"success": True,
"message": "pong",
"timestamp": frappe.utils.now()
}undefinedReturns:
服务器状态
Example:
GET /api/method/<app>.<module>.api.<endpoint>.ping
"""
return {
"success": True,
"message": "pong",
"timestamp": frappe.utils.now()
}undefinedhooks.pyundefinedhooks.pyundefinedundefinedundefined<app>/<module>/api/test_<endpoint_name>.py"""
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>.py"""
<端点名称> 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")undefinedundefined| 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 | 公开 | 健康检查 |
undefinedundefinedundefinedundefinedundefinedundefinedbench --site <site> run-tests --module "<app>.<module>.api.test_<endpoint_name>"bench --site <site> run-tests --module "<app>.<module>.api.test_<endpoint_name>"undefinedundefinedundefinedundefinedundefinedundefinedbench --site <site> run-tests --module <test_module>undefinedbench --site <site> run-tests --module <test_module>undefined_check_permission()frappe.db.commit()allow_guest=Truefrappe.throw()_check_permission()frappe.db.commit()allow_guest=Truefrappe.throw()