Attribute
查看 MarkdownAttribute 是属于一个 Flow run 的 durable 数据。Step 和 RPC 读写它。Dex 把它和 Flow 存在一起，所以中间状态不需要另外的 database。
Attribute 的生命周期和它的 Flow 相同。Flow 关闭后过了 retention period，Dex 删除 Flow，也一并删除 Attribute。要更长时间保留数据，见在 Flow 之后保留数据。
什么时候用 Attribute
Attribute 适合存放 Flow 据以做决定的状态：订单状态、计数器、客户资料，或者前一个 Step 的结果。Step 要等待并消费的消息，用 Channel。可以丢失的实时进度，用 Stream。
使用 Attribute
在 Flow 的 persistence schema 里声明每个 Attribute。之后 Step 或 RPC 就可以读写它。schema 没有声明的 Attribute，Worker 会拒绝。把 Attribute 设为 null 会删除它。
只有 Step 方法或 RPC 成功时，它的 Attribute 写入才会生效。见 Step 的保证。
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。它适合随数据增长的值，例如订单里每笔 payment 的状态。Dex 分别存储每个 key，所以一次写入不会重写整个集合。
progress = AttributeMap("primitive-attribute-progress", str)
progress.set(context, "payment", "authorized")
例子: examples/python/dex_examples/primitives/attribute/attribute_flow.py
读取和搜索 Attribute
Flow 外的代码通过 Client 读取 Attribute。什么时候改用 RPC，见什么时候读取需要 RPC。另见 Read Attributes from outside a Flow。在 Flow 启动时设置初始值，见 Initial values。
带 index 的 Attribute 可以用在 SearchFlows 查询里，查找 Flow run。上面的例子给 status Attribute 建了 index。index 类型、index key 和 AttributeMap index 见 Indexes。
Lock
两个 Step 或 RPC 可能同时写同一个 Attribute。没有 lock 时，两个先读后写这个 Attribute 的 Step 或 RPC 可能互相覆盖对方的更新，见 Step 的保证。用 Attribute lock 让它们依次执行。Dex 在 Step method 或 RPC 开始前获取 lock，结束后释放。Step 会等待它的 lock。RPC 拿不到 lock 就失败，由你重试。
lock 只对声明了它的 Step 和 RPC 起作用。没有声明这个 lock 的 Step 或 RPC 仍然可以写这个 Attribute。Step 按开始等待 lock 的顺序拿到它。RPC 不排队。即使有 Step 在等一个空闲的 lock，RPC 也会直接拿走它。所以持续不断的加锁 RPC 可能推迟这些 Step。让加锁的 RPC 尽量短。
这个 Step 在 Execute 运行期间锁住一个 singleton Attribute 和一个 AttributeMap instance：
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 的获取时机，以及 WaitFor、Execute 和 RPC 的 lock，见 Locks。lock conflict 见 RPC lock conflicts，锁住整张 AttributeMap 见 whole-map coordination。
短期 secret
hard-delete Attribute 用来保存短期 secret，例如 Flow 用几次就丢弃的 API key。删除它会永久清除存储的值。Dex API 和 Flow history 都无法恢复。和其他写入不同，hard-delete 写入在 Dex 保存 handler 的结果之前就生效。要求和后果见 Hard-delete Attributes。
在 Flow 之后保留数据
Attribute Store sync 把选中的 Attribute 复制到你自己的 PostgreSQL 表。每个目标称为一个 Attribute Store。Dex 删除 Flow 后，表里的数据仍然保留。你的应用和报表用 SQL 查询这些表，你也可以为自己的查询建 index。
同步一个 Attribute 的步骤：
- 在 server 启动前建好表。Dex 不会创建、迁移或回填表。
- 在 server 配置里配置这个 store。
- 用 syncToAttributeStore 标记 Attribute 或 AttributeMap。
- 启动 Flow 时，在它的 FlowConfig 里列出 store 名称。
需要知道的行为：
- Flow 仍是数据的权威来源。表里的一行保存最新值，不是历史记录。不要根据这张表做 Flow 决策。
- Dex 在后台写入。写入不会拖慢 Step、RPC 或 SetAttributes 调用。同一个 Flow ID 的写入保持顺序。
- 写入失败时，Dex 会在一段有限的时间内重试，之后丢弃这次写入并记录错误日志。Dex 不会重放旧值，也不会为新配置的 store 补写数据。
- Flow 自己完成或失败时，会保持运行，直到待写入的数据写完或被丢弃。StopFlow 不等待。
- 删除 Attribute 会把它的列写成 NULL。Flow 关闭时，Dex 不会删除这一行。
- 一个 store 可以把 Flow ID 拆成多个主键列，也可以把每个 AttributeMap instance 写成单独一张表里的一行。
配置、表的要求、row key、AttributeMap 表、重试、指标和本地开发，见英文的 Attribute Store sync 参考页。完整示例见 Entity Store 模式。
参考
每个 SDK 参考的 Attribute 页列出该语言的 API 名称。
Data primitive semantics 详细说明初始值、Client 读取、index、hard-delete Attribute 和 lock。