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, "keyword1"),
    )

    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。从 Flow 外读取 Attribute，见从 Flow 外读取 Attribute。Client 也可以用 WaitForAttributeMatch，等待 singleton Attribute 或某个 AttributeMap instance 满足 scalar 比较，并返回当前命中的值。见 Client。

从 Flow 外读取 Attribute

Flow 外的应用代码用 Client 的 GetAttributes 读取 Attribute 值。一次调用可以读取任意组合的 singleton Attribute 和 AttributeMap instance，所有值都来自同一次提交。Dex Server 自己响应读取，不会运行 RPC handler 或 Worker。

调用读取当前 run；没有打开的 run 时，读取最新 run。已关闭的 run 在 Dex 清除它之前仍可读取。调用只看到已提交的状态，而 RPC handler 还能看到它在同一次调用中先前的写入。Dex Server 在响应前解析 hard-delete 值；已删除的值和任何没有存储值的 Attribute 一样，读作未设置。

如果读取需要 Flow 内的 validation、authorization、派生视图或 Attribute lock，RPC 仍是合适的选择。示例、各 SDK 中未设置值的表示和错误，见 Client：读取 Attribute。

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 会消耗 namespace 级、数量有限的 Search Attribute 槽位，并且跨 Flow type 共用。

SDK 默认使用普通 Attribute 的名称作为 index key。应用索引必须显式覆盖这个默认值，使用带编号的通用 key。Attribute 本身保留业务名称，namespace 下的所有 Flow type 共用一组按类型规划的槽位。同一个 Flow 中不同的逻辑索引字段需要不同槽位；不同 Flow type 可以复用同一槽位，但索引类型必须一致，业务含义可以不同。Dex 系统索引保留其专用名称。

Index type	应用 index key
KEYWORD	keyword1、keyword2、…
KEYWORD_ARRAY	keywordList1、keywordList2、…
FULL_TEXT	text1、text2、…
INT	int1、int2、…
BOOL	bool1、bool2、…
DOUBLE	double1、double2、…
DATETIME	datetime1、datetime2、…

修改索引名不会为已有 Flow execution 回填新索引。示例 schema 的验证使用全新的本地 dexcli dev 实例；已有部署的迁移需单独安排。

同一个通用 key 在不同 Flow type 中可以有不同业务含义。应用搜索必须限定 FlowType。翻页时保持查询不变。Dex Web 已按所选 Flow type 限定搜索范围。将调用方条件放在括号内再与应用范围组合，避免 OR 绕过限制。不要索引敏感数据或 PII。Customer Refund 示例中的 customer-email 是合成数据。

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

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

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

FlowType = 'AttributeFlow' AND keyword1 = 'completed'

将这个 query 传给 SearchFlows：

page = await app_state.client.search_flows(
    "FlowType = 'AttributeFlow' AND keyword1 = '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：

FlowType = 'AttributeFlow' AND keyword2 = 'authorized'

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

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

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

Hard-delete Attribute

Hard-delete Attribute 适合保存短期使用的 secret，例如 Flow 只用几次、用完就丢弃的 API key。

无论 value 大小，Dex 都只把 hard-delete Attribute 的 value 存在对象存储中，Flow history 里只保留一个 reference。删除这个 Attribute，或把它设为 null，会永久删除存储的 value。之后，无论是 Dex API 还是 Flow history，都无法再恢复它。

从 server-zd v0.1.5 起，Flow run 关闭后、清除前，可以直接调用 FlowService 的 SetAttributes API 擦除 hard-delete Attribute。SDK Client 不提供这个 API，请用 dexcli api call SetAttributes 或其他 FlowService client 发送。将请求中每个 Attribute 都设为 null。每个 Attribute 必须仍存有 hard-delete value。擦除会永久删除存储的对象，run 仍保持关闭。对已关闭 run 的其他写入会返回 NOT_FOUND。

Step 和 RPC handler 会直接收到 value。SDK 不会把它写入本地 blob cache，Dex Server 也不会把它写入自己的 blob cache。

在定义上开启这个 option：Go 用 HardDeleteOnNull，Java 和 TypeScript 用 hardDeleteOnNull，Python 和 Rust 用 hard_delete_on_null。Attribute 和 AttributeMap 都支持。

class AttributeFlow(Flow[str]):
    model_api_key = Attribute(
        "primitive-attribute-model-api-key",
        str,
        hard_delete_on_null=True,
    )


class AttributeStep(Step[str]):
    def execute(self, context: Context, input: str) -> StepDecision:
        self.flow.status.set(context, "completed")
        # The Flow no longer needs the short-lived key, so delete it permanently.
        self.flow.model_api_key.delete(context)
        return graceful_complete(input)

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

Hard-delete Attribute 有以下要求：

只支持 string 和 object value（包括 bytes），不支持数值和 Boolean 类型。
不能设置 index，也不能同步到 Attribute Store。WaitForAttributeMatch 和 Action condition 都不接受它。
需要支持 hard-delete Attribute 的 server，例如 server-zd。SDK 不会向更早版本的 server 写入这个 Attribute。

每个 Flow 对每个 Attribute key 只保存一份 value，并在原位置覆盖。需要注意以下影响：

Step 或 RPC 的 attempt 在结果提交之前就会写入或删除存储的 value。因此，重试的 attempt、其他 Step 和 RPC，以及 Attribute 读取，都可能看到之后失败的 attempt 做的写入或删除。handler 应该可以安全地重复执行，不要依赖对这个 Attribute 的「先读再写」。
多个 Step 或 RPC 并发写同一个 key 时，最终的 value 不一定是最后提交的那次写入。
如果 Flow ID 已经在运行，Flow start 或 SubFlow start 会失败，但仍然会覆盖正在运行的 Flow 的 value；initial value 为 null 时会删除它。
复用同一个 Flow ID 的 Flow 共用这份存储的 value。
已删除的 value 总是读作不存在，time travel 也不会恢复它。重新启动的 SubFlow 从原来的 initial value 开始；如果之前的 run 已经删除了这个 value，新的 run 会读到不存在。
从定义上移除这个 option 后，已存储的 value 会一直保留，直到 Dex 删除这个 Flow。请先用 hard-delete 定义删除 value，再移除 option。

value 会一直保存，直到被删除，或者 Dex 在 retention period 结束后删除这个 Flow。Dex 之外的对象存储备份和 bucket versioning 可能会保留副本。参见对象存储。

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

首先，在一个 YAML 文件中定义 Attribute Store：

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

目前只支持 PostgreSQL 作为 Attribute Store。Redis 支持由 issue 616 跟踪。请先创建目标。 Attribute Sync 会异步写入最新状态，不会自动回填已有 Flow。目标契约、类型映射和重试行为请参阅自定义 Attribute Store 运维。

使用这个文件启动本地 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。

协调整张 AttributeMap 的写操作

AttributeMap 锁只保护一个指定 instance。若一个写入约束涉及多个 instance 或整个集合，可注册另外一个普通 bool Attribute 作为协调锁。例如，在同一个 PersistenceSchema 中注册 AttributeMap currentMessages，以及非 map、非索引的 bool Attribute currentMessagesLock。

所有写入、删除、归档或以其他方式修改受保护 map 的 Step、timeout handler 和 RPC，都必须在对应方法的锁选项中请求同一个普通 Attribute 锁。RPC 在注册时声明这把固定锁。只获取 map instance 锁的写入方不会与该普通 Attribute 锁冲突。参与者在同一个 Flow execution 内串行执行，其他 Flow execution 拥有自己的锁。

bool 值不表示锁状态，不要通过读取或切换 true/false 获取或释放锁。Dex 在 invocation 周围管理锁的获取和释放，协调 Attribute 无需持久化一个值。锁也不会加载 map 数据；仍须声明所读取的 instance，只有必须枚举时才加载整个 map。RPC 获取锁时可能返回锁冲突错误，而非让调用方排队等待。

独立分区的更新保留 instance 锁即可。需要维护整个集合的写入约束时，所有相关写入方使用这把共享的普通 Attribute 锁。

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

使用 Attribute​

AttributeMap​

Initial Attribute values​

从 Flow 外读取 Attribute​

Indexed Attribute​

Hard-delete Attribute​

Attribute Sync​

Locking​

协调整张 AttributeMap 的写操作​

Lock WaitFor​

Lock Execute​

Lock an RPC​

使用 Attribute

AttributeMap

Initial Attribute values

从 Flow 外读取 Attribute

Indexed Attribute

Hard-delete Attribute

Attribute Sync

Locking

协调整张 AttributeMap 的写操作

Lock WaitFor

Lock Execute

Lock an RPC