Attribute

一个 Attribute 让你可以在一个 Flow execution 内存储数据，作为 persistence storage。

这尤其适合在 Flow execution 期间存储中间数据，省去设置和管理 database dependency 的工作。

Dex Attribute 不是用来替代 database 的。一个明显的限制是，Attribute 只在它所属的 Flow execution 生命周期内存在。Dex 会在 Flow 关闭后的 configurable retention period 结束后删除 Flow execution 时，一并删除 Attribute。

不过，Dex 允许你将 Attribute values 同步到 application database，作为 permanent storage。这也让你可以构建更强大的 database indexes 来查询 Attribute。

使用 Attribute

将 Attribute 定义为 Flow persistence schema 的一部分。

之后就可以在 Step 或 RPC implementation 内读写 Attribute。Attribute 必须先通过 PersistenceSchema 声明，否则 Worker 会拒绝这些操作。

更新 Attribute 时，将它设为 null 会删除它。

class AttributeStep(Step[str]):
    def __init__(self, flow: "AttributeFlow") -> None:
        self.flow = flow

    def wait_for(self, context: Context, input: str) -> Wait:
        self.flow.status.set(context, "processing")
        return Wait.skip_immediately()

    def execute(self, context: Context, input: str) -> StepDecision:
        self.flow.status.set(context, "completed")
        return graceful_complete(input)


class AttributeFlow(Flow[str]):
    status = Attribute(
        "primitive-attribute-status",
        str,
        index=AttributeIndex(IndexType.KEYWORD, "OrderStatus"),
    )

    def __init__(self) -> None:
        self.start = AttributeStep(self)

    def get_steps(self) -> StepList[str]:
        return StepList.start_step(self.start)

    def get_persistence_schema(self) -> PersistenceSchema:
        return PersistenceSchema.of(self.status)

例子: examples/python/dex_examples/primitives/attribute/attribute_flow.py

AttributeMap

AttributeMap 是一组 key 在运行时才确定的 Attribute。它适合 Flow 需要一组不断变化的小值的场景，例如订单中每笔 payment 或每个 item 的 status。

AttributeMap 的原因是为了高效存储一组数据。每个 Attribute 在 Dex Server 中都作为一个 blob 存储。把所有数据放进一个 Attribute，意味着每次更新都会重写整个 blob。AttributeMap 让 Dex Server 将每个 instance 作为独立 blob 存储，从而获得更高效率。

progress = AttributeMap("primitive-attribute-progress", str)

progress.set(context, "payment", "authorized")

例子: examples/python/dex_examples/primitives/attribute/attribute_flow.py

Initial Attribute values

StartFlowOptions 可以在创建 Flow 时设置初始 Attribute。初始值与创建 Flow 是一个操作，所以第一个 Step 立刻能看到它们。见 Flow options。

Flow 启动后，应用通过 Flow RPC 读写 Attribute。Client 也可以用 WaitForAttributeMatch，等待 singleton Attribute 或某个 AttributeMap instance 满足 scalar 比较，并返回当前命中的值。见 Client。

Indexed Attribute

Dex 让你可以为 Attribute 配置 index。有了 indexing，就可以在 SearchFlows query 中使用这个 field，找出匹配 query 的 Flow execution。

Index type	值
KEYWORD	一个精确字符串。
FULL_TEXT	用于 tokenized search 的文本。
KEYWORD_ARRAY	一组精确字符串。
INT、DOUBLE、BOOL	对应的 scalar value。
DATETIME	一个日期与时间值。

index key 是 search schema 中的 key，也是 query 中使用的 key。index key 跨 Flow type 共用。

普通 Attribute 默认用 Attribute name 作为 index key。你也可以设置显式 index key。

status = Attribute(
    "primitive-attribute-status",
    str,
    index=AttributeIndex(IndexType.KEYWORD, "OrderStatus"),
)

例子: examples/python/dex_examples/primitives/attribute/attribute_flow.py

上面的 status Attribute 使用 OrderStatus 作为 index key。要找 status 为 completed 的 Flow execution，可以使用这个 query：

OrderStatus = 'completed'

将这个 query 传给 SearchFlows：

page = await app_state.client.search_flows(
    "OrderStatus = 'completed'",
    20,
    "",
)

例子: examples/python/dex_examples/primitives/client-apis/controller.py

带 index 的 AttributeMap 必须设置显式 index key。它不是所有 Map instance 的 index。每个 instance 的 write 都更新同一个 search field，所以一个 instance 可以替换另一个 instance 写入的值。Map instance key 不能被搜索。

这不会把 Map 自动变成 KEYWORD_ARRAY。KEYWORD_ARRAY 只会 index 一次 write 传入的 string array；Dex 不会合并所有 Map instance 的值。下面的例子写入 payment instance 的值 authorized 后，可以用下面的 query 查询当前 search value：

OrderProgress = 'authorized'

如果之后另一个 Map instance 写入不同的值，它会替换 OrderProgress 的 search value。需要搜索一组值时，应使用带 KEYWORD_ARRAY 的普通 Attribute。不要用 AttributeMap index 查找所有有某个值的 instance。

progress = AttributeMap(
    "primitive-attribute-progress",
    str,
    index=AttributeIndex(IndexType.KEYWORD, "OrderProgress"),
)

例子: examples/python/dex_examples/primitives/attribute/attribute_flow.py

Attribute Sync

Dex Attribute 不是用来替代 database 的。Attribute Sync feature 会把 Attribute values 同步到 application database，无需你写一行代码。

这样就可以把 Attributes 作为 permanent storage 存到 database 中。这也让你可以构建更强大的 database indexes 来查询 Attribute。

标记每个需要同步的 Attribute：

display_name = Attribute("display_name", str, sync_to_attribute_store=True)

例子: examples/python/dex_examples/patterns/entity-store/user_profile_flow.py

首先，在 Dex Server 配置中定义 Attribute Store：

attributeStore:
  stores:
    entityStore:
      type: postgres
      dsn: postgres://entity_store:entity_store@localhost:55432/entity_store?sslmode=disable
      tableName: public.user_profiles

使用这个配置启动 Dex Server：

dexcli dev --attribute-store-config ./attribute-store.yaml

然后，在启动 Flow 时选择同一个 store name：

options = StartFlowOptions(
    config_override=FlowConfig(attribute_store_names=["entityStore"]),
)
await client.start_flow(flow, user_id, None, options)

例子: examples/python/dex_examples/patterns/entity-store/controller.py

Locking

Concurrent Step 和 RPC invocation 可以读取和写入同一个 Attribute。这会导致 race condition。为避免这种情况，请使用 Attribute lock。Dex Server 确保同一时间只有一个 Step execution 或 RPC handling 可以获取 lock。

Lock WaitFor

def get_step_options(self) -> StepOptions:
    return StepOptions(
        wait_for_lock_attributes=(
            self.flow.status.lock(),
            self.flow.progress.lock("payment"),
        ),
    )

例子: examples/python/dex_examples/primitives/attribute/attribute_flow.py

Lock Execute

def get_step_options(self) -> StepOptions:
    return StepOptions(
        execute_lock_attributes=(
            self.flow.status.lock(),
            self.flow.progress.lock("payment"),
        ),
    )

例子: examples/python/dex_examples/primitives/attribute/attribute_flow.py

Lock an RPC

@rpc(lock_attributes=(status.lock(), progress.lock("payment")))
def update_status(self, context: Context, input: str) -> RPCResult[str]:
    self.status.set(context, input)
    self.progress.set(context, "payment", input)
    return RPCResult(input)

例子: examples/python/dex_examples/primitives/attribute/attribute_flow.py