“

系列：S12 开发者教程 · 第 4 篇 | 难度：入门 | 阅读时间：15 分钟

第一个 Action：定义并执行业务操作

#引言

在前三篇教程中，我们学习了部署 coomia-dip、创建 Ontology 模型以及读写数据。但在真实的企业应用中，数据操作往往不是简单的 CRUD，而是带有业务规则的复合操作。例如"员工入职"涉及创建员工记录、分配部门、开通权限、发送通知等多个步骤。coomia-dip 中的 Action 正是用来封装这类业务操作的核心抽象。

Action 是 coomia-dip 对标 Palantir Foundry 中 Action Type 的实现。它允许你定义带有参数、校验规则和副作用的业务操作，并通过 SDK 或 API 统一调用。Action 的执行是事务性的——要么全部成功，要么全部回滚。

#核心概念

#什么是 Action？

Action 是一个可执行的业务操作定义，包含以下要素：

名称和描述：标识这个操作做什么
参数（Parameters）：操作所需的输入
校验规则（Validations）：执行前的业务规则检查
操作步骤（Operations）：实际执行的数据变更
副作用（Side Effects）：操作完成后的通知、日志等

Code

Action
├── name: string              # 如 "onboard_employee"
├── display_name: string      # 如 "员工入职"
├── parameters: Parameter[]   # 输入参数列表
│   ├── name: string
│   ├── type: PropertyType
│   ├── required: boolean
│   └── default: any
├── validations: Validation[] # 校验规则
├── operations: Operation[]   # 数据操作
└── side_effects: Effect[]    # 副作用

#Action vs 直接 CRUD

维度	直接 CRUD	Action
业务语义	无	有明确的业务含义
参数校验	手动	声明式校验规则
事务性	需手动管理	自动事务
审计日志	需手动记录	自动记录
权限控制	基于数据	基于操作
可发现性	无	API 自动暴露

#创建第一个 Action

#场景：员工入职

Python

from ontology_sdk import OntoPlatform

platform = OntoPlatform(
    control_plane_url="localhost:50051",
    data_plane_url="localhost:50052"
)

# 定义"员工入职"Action
onboard_action = platform.actions.create(
    name="onboard_employee",
    display_name="员工入职",
    description="新员工入职流程：创建员工记录、分配部门、设置初始权限",
    parameters={
        "name": {
            "type": "string",
            "required": True,
            "description": "员工姓名"
        },
        "email": {
            "type": "string",
            "required": True,
            "description": "工作邮箱",
            "constraints": {"format": "email"}
        },
        "department_id": {
            "type": "string",
            "required": True,
            "description": "入职部门 ID"
        },
        "title": {
            "type": "string",
            "required": True,
            "description": "职位名称"
        },
        "level": {
            "type": "integer",
            "required": True,
            "description": "职级",
            "constraints": {"min": 1, "max": 15}
        },
        "salary": {
            "type": "decimal",
            "required": True,
            "description": "月薪"
        },
        "hire_date": {
            "type": "date",
            "default": "today",
            "description": "入职日期，默认今天"
        }
    },
    validations=[
        {
            "name": "check_department_exists",
            "type": "object_exists",
            "object_type": "Department",
            "primary_key": {"ref": "parameters.department_id"},
            "error_message": "指定的部门不存在"
        },
        {
            "name": "check_email_unique",
            "type": "unique_check",
            "object_type": "Employee",
            "property": "email",
            "value": {"ref": "parameters.email"},
            "error_message": "邮箱已被使用"
        },
        {
            "name": "check_headcount",
            "type": "custom_function",
            "function": "validate_headcount",
            "args": {"department_id": {"ref": "parameters.department_id"}},
            "error_message": "部门编制已满"
        }
    ],
    operations=[
        {
            "type": "create_object",
            "object_type": "Employee",
            "properties": {
                "employee_id": {"ref": "generated.employee_id"},
                "name": {"ref": "parameters.name"},
                "email": {"ref": "parameters.email"},
                "title": {"ref": "parameters.title"},
                "level": {"ref": "parameters.level"},
                "salary": {"ref": "parameters.salary"},
                "hire_date": {"ref": "parameters.hire_date"},
                "is_active": True
            },
            "output_ref": "new_employee"
        },
        {
            "type": "create_relation",
            "relation_type": "belongs_to",
            "source": {"ref": "new_employee.rid"},
            "target": {"ref": "parameters.department_id"},
            "properties": {
                "joined_at": {"ref": "parameters.hire_date"},
                "role_in_dept": "member"
            }
        }
    ],
    side_effects=[
        {
            "type": "send_notification",
            "channel": "email",
            "template": "welcome_employee",
            "to": {"ref": "parameters.email"},
            "data": {
                "name": {"ref": "parameters.name"},
                "department": {"ref": "department.name"}
            }
        },
        {
            "type": "audit_log",
            "event": "employee_onboarded",
            "data": {
                "employee_id": {"ref": "new_employee.properties.employee_id"},
                "department_id": {"ref": "parameters.department_id"}
            }
        }
    ]
)

print(f"Action 创建成功: {onboard_action.name}")
print(f"  参数: {', '.join(p.name for p in onboard_action.parameters)}")

#执行 Action

#基本执行

Python

# 执行员工入职 Action
result = platform.actions.execute(
    action_name="onboard_employee",
    parameters={
        "name": "新员工A",
        "email": "newemployee@example.com",
        "department_id": "DEPT-001",
        "title": "初级工程师",
        "level": 3,
        "salary": 15000.00
    }
)

print(f"执行结果: {result.status}")
print(f"  员工 RID: {result.outputs['new_employee']['rid']}")
print(f"  员工编号: {result.outputs['new_employee']['properties']['employee_id']}")

#执行结果

Action 执行后返回一个结果对象：

Python

ActionResult(
    status="SUCCESS",           # SUCCESS, FAILED, VALIDATION_ERROR
    action_name="onboard_employee",
    execution_id="exec-xxx",    # 执行唯一标识
    outputs={                   # 操作输出
        "new_employee": {
            "rid": "ri.ontology.object.employee.xxx",
            "properties": {"employee_id": "EMP-000006", ...}
        }
    },
    validation_errors=[],       # 校验错误列表
    execution_time_ms=234,      # 执行耗时
    audit_trail={               # 审计信息
        "user": "admin",
        "timestamp": "2024-03-24T10:00:00Z",
        "ip": "192.168.1.100"
    }
)

#处理校验错误

Python

# 尝试用重复邮箱入职
result = platform.actions.execute(
    action_name="onboard_employee",
    parameters={
        "name": "重复员工",
        "email": "newemployee@example.com",  # 已被使用
        "department_id": "DEPT-001",
        "title": "工程师",
        "level": 3,
        "salary": 15000.00
    }
)

if result.status == "VALIDATION_ERROR":
    for error in result.validation_errors:
        print(f"校验失败: {error.validation_name} - {error.message}")
        # 输出: 校验失败: check_email_unique - 邮箱已被使用

#更多 Action 示例

#员工调岗

Python

transfer_action = platform.actions.create(
    name="transfer_employee",
    display_name="员工调岗",
    description="将员工从一个部门调至另一个部门",
    parameters={
        "employee_id": {"type": "string", "required": True},
        "new_department_id": {"type": "string", "required": True},
        "new_title": {"type": "string"},
        "effective_date": {"type": "date", "default": "today"}
    },
    validations=[
        {
            "name": "check_employee_active",
            "type": "property_check",
            "object_type": "Employee",
            "primary_key": {"ref": "parameters.employee_id"},
            "property": "is_active",
            "expected": True,
            "error_message": "员工已离职，无法调岗"
        }
    ],
    operations=[
        {
            "type": "delete_relation",
            "relation_type": "belongs_to",
            "source_filter": {"employee_id": {"ref": "parameters.employee_id"}}
        },
        {
            "type": "create_relation",
            "relation_type": "belongs_to",
            "source": {"ref": "parameters.employee_id"},
            "target": {"ref": "parameters.new_department_id"},
            "properties": {
                "joined_at": {"ref": "parameters.effective_date"}
            }
        },
        {
            "type": "update_object",
            "object_type": "Employee",
            "primary_key": {"ref": "parameters.employee_id"},
            "properties": {
                "title": {"ref": "parameters.new_title"}
            },
            "condition": {"ref": "parameters.new_title", "is_not_null": True}
        }
    ]
)

#项目状态变更

Python

change_status_action = platform.actions.create(
    name="change_project_status",
    display_name="变更项目状态",
    parameters={
        "project_id": {"type": "string", "required": True},
        "new_status": {
            "type": "enum",
            "values": ["planning", "in_progress", "on_hold", "completed", "cancelled"],
            "required": True
        },
        "reason": {"type": "string"}
    },
    validations=[
        {
            "name": "check_valid_transition",
            "type": "custom_function",
            "function": "validate_status_transition",
            "args": {
                "project_id": {"ref": "parameters.project_id"},
                "new_status": {"ref": "parameters.new_status"}
            },
            "error_message": "不允许的状态转换"
        }
    ],
    operations=[
        {
            "type": "update_object",
            "object_type": "Project",
            "primary_key": {"ref": "parameters.project_id"},
            "properties": {
                "status": {"ref": "parameters.new_status"}
            }
        }
    ],
    side_effects=[
        {
            "type": "send_notification",
            "channel": "webhook",
            "url": "https://hooks.example.com/project-updates",
            "data": {
                "project_id": {"ref": "parameters.project_id"},
                "new_status": {"ref": "parameters.new_status"},
                "reason": {"ref": "parameters.reason"}
            }
        }
    ]
)

#Action 权限控制

Python

# 为 Action 设置执行权限
platform.actions.set_permissions(
    action_name="onboard_employee",
    permissions={
        "allowed_roles": ["hr_manager", "admin"],
        "allowed_groups": ["human-resources"],
        "require_approval": False
    }
)

# 需要审批的 Action
platform.actions.set_permissions(
    action_name="transfer_employee",
    permissions={
        "allowed_roles": ["hr_manager", "department_head"],
        "require_approval": True,
        "approval_chain": ["department_head", "hr_director"]
    }
)

#Action 执行历史

Python

# 查看 Action 执行历史
history = platform.actions.get_execution_history(
    action_name="onboard_employee",
    limit=20
)

for exec_record in history:
    print(f"  {exec_record.execution_id}")
    print(f"    时间: {exec_record.timestamp}")
    print(f"    用户: {exec_record.user}")
    print(f"    状态: {exec_record.status}")
    print(f"    耗时: {exec_record.execution_time_ms}ms")

#使用 YAML 定义 Action

除了 SDK 编程方式，coomia-dip 还支持通过 YAML 文件声明式地定义 Action：

YAML

# actions/onboard_employee.yaml
name: onboard_employee
display_name: 员工入职
description: 新员工入职流程

parameters:
  - name: name
    type: string
    required: true
    description: 员工姓名
  - name: email
    type: string
    required: true
    constraints:
      format: email
  - name: department_id
    type: string
    required: true
  - name: title
    type: string
    required: true
  - name: level
    type: integer
    required: true
    constraints:
      min: 1
      max: 15
  - name: salary
    type: decimal
    required: true

validations:
  - name: check_department_exists
    type: object_exists
    object_type: Department
    primary_key: $parameters.department_id
  - name: check_email_unique
    type: unique_check
    object_type: Employee
    property: email
    value: $parameters.email

operations:
  - type: create_object
    object_type: Employee
    properties:
      employee_id: $generated.employee_id
      name: $parameters.name
      email: $parameters.email
      title: $parameters.title
      level: $parameters.level
      salary: $parameters.salary
      is_active: true
    output_ref: new_employee
  - type: create_relation
    relation_type: belongs_to
    source: $new_employee.rid
    target: $parameters.department_id

permissions:
  allowed_roles: [hr_manager, admin]

Python

# 从 YAML 加载 Action
platform.actions.load_from_yaml("actions/onboard_employee.yaml")

#与 Palantir Foundry 对比

特性	Palantir Foundry	coomia-dip
Action 定义	Web UI / TypeScript	SDK + YAML + API
参数校验	内置校验	声明式 + 自定义函数
事务支持	是	是
审批流程	Workshop	内置审批链
审计日志	自动	自动
权限控制	RBAC	RBAC + ABAC
副作用	Webhooks	多通道通知 + Webhook

#最佳实践

原子性设计：一个 Action 应该代表一个完整的业务操作
充分校验：在 operations 之前进行所有必要的校验
明确参数：每个参数都要有类型、描述和必要的约束
副作用分离：通知、日志等放在 side_effects 中，不影响核心操作
权限最小化：只授予必要的角色执行权限
版本管理：将 YAML Action 定义纳入版本控制

#总结

在本教程中，我们学习了 coomia-dip 的 Action 机制：

Action 概念：封装业务操作的核心抽象，包含参数、校验、操作和副作用
创建 Action：通过 SDK 或 YAML 定义 Action
执行 Action：调用 Action 并处理结果和错误
权限和审计：为 Action 配置权限控制和审计日志

Action 是连接数据模型和业务流程的桥梁。通过 Action，你可以将复杂的业务逻辑封装为可复用、可审计、可控制的操作单元。

本文是 coomia-dip 开发者教程系列的第 4 篇。 项目地址：https://github.com/coomia-dip/coomia-dip↗ | 许可证：Apache License 2.0