frappe-api

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

Frappe 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 --public

Procedure

操作步骤

Step 1: Gather API Requirements

步骤1:收集API需求

Ask the user for:
  1. Endpoint Name (snake_case, e.g.,
    get_dashboard_stats
    )
  2. HTTP Methods supported (GET, POST, PUT, DELETE)
  3. Authentication Type:
    • Token (API Key + Secret)
    • Session (Cookie-based)
    • OAuth 2.0
    • Public (no auth required - use sparingly)
  4. Parameters - Input parameters with types
  5. Related DocType (if applicable)
  6. Allowed Roles (who can access this endpoint)
向用户确认以下信息:
  1. 端点名称(蛇形命名,例如:
    get_dashboard_stats
  2. 支持的HTTP方法(GET、POST、PUT、DELETE)
  3. 认证类型
    • Token(API密钥+密钥密码)
    • Session(基于Cookie)
    • OAuth 2.0
    • 公开(无需认证 - 谨慎使用)
  4. 参数 - 带类型的输入参数
  5. 关联DocType(如适用)
  6. 允许的角色(哪些用户可访问该端点)

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>.py
:
python
"""
<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
python
"""
<端点名称> 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
,传入非整数会自动触发校验错误。

Python type hints. For example, if you declare
limit: int
, passing

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()
}
undefined

Step 4: Generate API v2 Endpoints (Optional)

步骤4:生成v2 API端点(可选)

For v2 REST API pattern, create custom routes in
hooks.py
:
python
undefined
如需v2 REST API模式,在
hooks.py
中创建自定义路由:
python
undefined

hooks.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"}, ]
undefined
website_route_rules = [ {"from_route": "/api/v2/<app>/<endpoint>", "to_route": "<app>.<module>.api.<endpoint>.handle_v2"}, ]
undefined

Step 5: Generate API Tests

步骤5:生成API测试

Create
<app>/<module>/api/test_<endpoint_name>.py
:
python
"""
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
python
"""
<端点名称> 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文档预览

undefined
undefined

API 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:

端点列表:

MethodEndpointAuthDescription
GET.getToken/SessionGet single document
GET.get_listToken/SessionList with pagination
POST.createToken/SessionCreate document
PUT.updateToken/SessionUpdate document
DELETE.deleteToken/SessionDelete document
POST.submitToken/SessionSubmit for processing
POST.cancelToken/SessionCancel document
POST.bulk_update_statusToken/SessionBulk status update
GET.pingPublicHealth check
方法端点认证方式描述
GET.getToken/Session获取单个文档
GET.get_listToken/Session带分页的列表查询
POST.createToken/Session创建文档
PUT.updateToken/Session更新文档
DELETE.deleteToken/Session删除文档
POST.submitToken/Session提交文档进行处理
POST.cancelToken/Session取消文档
POST.bulk_update_statusToken/Session批量更新状态
GET.ping公开健康检查

Authentication:

认证方式:

bash
undefined
bash
undefined

Token auth (recommended for integrations)

Token认证(推荐用于集成场景)

curl -H "Authorization: token api_key:api_secret"
https://site.com/api/method/<endpoint>
curl -H "Authorization: token api_key:api_secret"
https://site.com/api/method/<endpoint>

Session auth (for browser clients)

Session认证(适用于浏览器客户端)

First login, then use session cookie

先登录,然后使用会话Cookie

undefined
undefined

Files 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模块?
undefined

Step 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

输出格式

undefined
undefined

API 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
undefined
bash
undefined

Get document

获取文档

curl -X GET "https://site.com/api/method/<app>.<module>.api.<endpoint>.get?name=DOC-001"
-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"

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"}'
undefined
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"}'
undefined

Next Steps:

后续步骤:

  1. Create API keys: Setup > User > API Access
  2. Test endpoints with curl or Postman
  3. Run API tests:
    bench --site <site> run-tests --module <test_module>
undefined
  1. 创建API密钥:设置 > 用户 > API访问
  2. 使用curl或Postman测试端点
  3. 运行API测试:
    bench --site <site> run-tests --module <test_module>
undefined

Rules

规则

  1. Always Check Permissions — Every endpoint must call
    _check_permission()
    first
  2. Validate All Input — Never trust user input, validate and sanitize everything
  3. Type Annotations — Use Python type hints for v15 auto-validation
  4. Transaction Handling — Frappe auto-commits on successful requests; manual
    frappe.db.commit()
    rarely needed except in background jobs
  5. Public Endpoints — Use
    allow_guest=True
    sparingly, only for truly public data
  6. Error Handling — Use
    frappe.throw()
    with appropriate exception types
  7. Documentation — Every endpoint must have docstring with Args/Returns/Example
  8. ALWAYS Confirm — Never create files without explicit user approval
  1. 始终检查权限 — 每个端点必须首先调用
    _check_permission()
  2. 校验所有输入 — 绝不信任用户输入,对所有输入进行校验和清理
  3. 类型注解 — 使用Python类型提示以利用v15的自动校验功能
  4. 事务处理 — Frappe会在请求成功时自动提交;除后台任务外,极少需要手动调用
    frappe.db.commit()
  5. 公开端点 — 仅在处理真正公开的数据时才使用
    allow_guest=True
    ,务必谨慎
  6. 错误处理 — 使用
    frappe.throw()
    并传入合适的异常类型
  7. 文档完善 — 每个端点必须包含带Args/Returns/Example的文档字符串
  8. 必须确认 — 未经用户明确确认,绝不创建文件