Python SDK

Ice Python SDK (ice-rules), with full feature parity with the Java/Go SDKs. Supports decorator-based leaf node registration and asyncio-based async client.

Installation

pip install ice-rules

Quick Start

import ice

# Register leaf node with decorator
@ice.leaf("com.example.ScoreFlow", name="Score Check")
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

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

# Execute rule
pack = ice.Pack(ice_id=1)
pack.roam.put("score", 85.0)
result = ice.sync_process(pack)

Leaf Node Development

Decorator Registration

@ice.leaf(class_name, name=None, desc=None, order=0, alias=None)
  • class_name: Node class name, corresponding to confName in Server configuration
  • name: Node display name
  • desc: Node description
  • alias: List of aliases for compatibility with class names configured by other language SDKs

The SDK automatically detects the node type based on the implemented method:

TypeMethodReturn ValueDescription
Flowdo_roam_flow(roam)boolConditional check (most common)
do_pack_flow(pack)boolNeeds Pack access
do_flow(ice_ctx)boolNeeds full Context
Resultdo_roam_result(roam)boolBusiness operation (most common)
do_pack_result(pack)boolNeeds Pack access
do_result(ice_ctx)boolNeeds full Context
Nonedo_roam_none(roam)NoneAuxiliary operation (most common)
do_pack_none(pack)NoneNeeds Pack access
do_none(ice_ctx)NoneNeeds full Context

Field Declaration

Use Annotated type hints to add field descriptions:

from typing import Annotated

@ice.leaf("com.example.AmountResult", name="Amount Dispenser")
class AmountResult:
    key: Annotated[str, ice.IceField(name="User Key", desc="Key for user ID in roam")] = ""
    value: Annotated[float, ice.IceField(name="Dispense Amount")] = 0.0

    # Hidden from configuration interface
    _internal: str = ""  # Underscore prefix is automatically ignored

    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 is also supported for explicitly ignoring fields.

Lifecycle Hooks

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

    def after_properties_set(self):
        """Automatically called after configuration is loaded/updated"""
        self.compiled = compile_expression(self.exp)

Executing Rules

Synchronous Execution

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

# Multiple calling styles
ctx_list = ice.sync_process(pack)
roam = ice.process_single_roam(pack)
roams = ice.process_roam(pack)
ctx = ice.process_single_ctx(pack)

Asynchronous Execution

import asyncio

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

Trigger Methods

pack = ice.Pack(ice_id=1)      # By iceId
pack = ice.Pack(scene="recharge")  # By scene
pack = ice.Pack(conf_id=123)   # By node ID

Priority: ice_id > scene > conf_id.

Roam Operations

Roam is based on threading.RLock and is thread-safe:

roam = ice.Roam()

# Basic operations
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)  # Supports "true"/"1"/"yes" strings

# Multi-level key
roam.put_multi("user.profile.level", 5)
level = roam.get_multi("user.profile.level")

# Reference syntax
roam.get_union("@user.profile.level")  # 5

Client Configuration

Synchronous Client

client = ice.FileClient(
    app=1,
    storage_path="./ice-data",
    poll_interval=5,          # Version poll interval (seconds), default 5
    heartbeat_interval=30,    # Heartbeat interval (seconds), default 30
)
client.start()
client.wait_started()
client.destroy()

Asynchronous Client

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

AsyncFileClient uses asyncio.create_task() to manage background tasks, with file I/O executed via asyncio.to_thread().

Error Handling

Global Error Handler

def my_error_handler(node, ctx, error):
    print(f"Node {node.ice_node_id} execution error: {error}")
    return ice.RunState.NONE  # Continue execution

ice.set_global_error_handler(my_error_handler)

Node-Level Error Handling

@ice.leaf("com.example.SafeNode")
class SafeNode:
    def do_roam_result(self, roam):
        # Business logic
        pass

    def error_handle(self, ctx, error):
        return ice.RunState.NONE  # Ignore error, continue execution

Next Steps