Python SDK

Ice Python SDK（ice-rules），功能与 Java/Go SDK 对等。支持装饰器注册叶子节点和 asyncio 异步客户端。

安装

pip install ice-rules

快速开始

import ice

# 装饰器注册叶子节点
@ice.leaf("com.example.ScoreFlow", name="分数判断")
class ScoreFlow:
    score: float = 0.0
    key: str = "score"

    def do_roam_flow(self, roam):
        value = roam.get_float(self.key, 0.0)
        return value >= self.score

# 启动 Client
client = ice.FileClient(app=1, storage_path="./ice-data")
client.start()

# 执行规则
pack = ice.Pack(ice_id=1)
pack.roam.put("score", 85.0)
result = ice.sync_process(pack)

叶子节点开发

装饰器注册

@ice.leaf(class_name, name=None, desc=None, order=0, alias=None)
  • class_name：节点类名，与 Server 配置中的 confName 对应
  • name：节点显示名称
  • desc：节点描述
  • alias：别名列表，兼容其他语言 SDK 配置的类名

SDK 根据实现的方法自动检测节点类型：

类型方法返回值说明
Flowdo_roam_flow(roam)bool条件判断（最常用）
do_pack_flow(pack)bool需要访问 Pack
do_flow(ice_ctx)bool需要完整 Context
Resultdo_roam_result(roam)bool业务操作（最常用）
do_pack_result(pack)bool需要访问 Pack
do_result(ice_ctx)bool需要完整 Context
Nonedo_roam_none(roam)None辅助操作（最常用）
do_pack_none(pack)None需要访问 Pack
do_none(ice_ctx)None需要完整 Context

字段声明

使用 Annotated 类型注解添加字段描述：

from typing import Annotated

@ice.leaf("com.example.AmountResult", name="金额发放")
class AmountResult:
    key: Annotated[str, ice.IceField(name="用户Key", desc="roam中用户ID的key")] = ""
    value: Annotated[float, ice.IceField(name="发放金额")] = 0.0

    # 不在配置界面显示
    _internal: str = ""  # 下划线前缀自动忽略

    def do_roam_result(self, roam):
        uid = roam.get_int(self.key, 0)
        if uid <= 0 or self.value <= 0:
            return False
        return send_service.send_amount(uid, self.value)

也支持 @ice.IceIgnore 显式忽略字段。

生命周期钩子

@ice.leaf("com.example.MyNode")
class MyNode:
    exp: str = ""

    def after_properties_set(self):
        """配置加载/更新后自动调用"""
        self.compiled = compile_expression(self.exp)

执行规则

同步执行

pack = ice.Pack(ice_id=1)
pack.roam.put("uid", 12345)

# 多种调用方式
ctx_list = ice.sync_process(pack)
roam = ice.process_single_roam(pack)
roams = ice.process_roam(pack)
ctx = ice.process_single_ctx(pack)

异步执行

import asyncio

pack = ice.Pack(ice_id=1)
pack.roam.put("uid", 12345)
ctx_list = await ice.async_process(pack)

触发方式

pack = ice.Pack(ice_id=1)      # 按 iceId
pack = ice.Pack(scene="recharge")  # 按场景
pack = ice.Pack(conf_id=123)   # 按节点 ID

优先级：ice_id > scene > conf_id。

Roam 操作

Roam 基于 threading.RLock，线程安全：

roam = ice.Roam()

# 基础操作
roam.put("name", "Alice")
name = roam.get_str("name")
age = roam.get_int("age", 0)
score = roam.get_float("score", 0.0)
flag = roam.get_bool("flag", False)  # 支持 "true"/"1"/"yes" 字符串

# 多级 key
roam.put_multi("user.profile.level", 5)
level = roam.get_multi("user.profile.level")

# 引用语法
roam.get_union("@user.profile.level")  # 5

Client 配置

同步客户端

client = ice.FileClient(
    app=1,
    storage_path="./ice-data",
    poll_interval=5,          # 版本轮询间隔（秒），默认 5
    heartbeat_interval=30,    # 心跳间隔（秒），默认 30
)
client.start()
client.wait_started()
client.destroy()

异步客户端

client = ice.AsyncFileClient(
    app=1,
    storage_path="./ice-data",
)
await client.start()
await client.wait_started()
await client.destroy()

AsyncFileClient 使用 asyncio.create_task() 管理后台任务，文件 I/O 通过 asyncio.to_thread() 执行。

错误处理

全局错误处理

def my_error_handler(node, ctx, error):
    print(f"节点 {node.ice_node_id} 执行出错: {error}")
    return ice.RunState.NONE  # 继续执行

ice.set_global_error_handler(my_error_handler)

节点级错误处理

@ice.leaf("com.example.SafeNode")
class SafeNode:
    def do_roam_result(self, roam):
        # 业务逻辑
        pass

    def error_handle(self, ctx, error):
        return ice.RunState.NONE  # 忽略错误，继续执行

下一步