---
title: "Attribute"
description: "Attribute 是属于一个 Flow run 的 durable 数据。Step 和 RPC 读写它。Dex 把它和 Flow 存在一起，所以中间状态不需要另外的 database。"
canonical_url: "https://docs.superdurable.io/zh-Hans/concepts/primitives/attribute/"
source: "docs/i18n/zh-Hans/docusaurus-plugin-content-docs/current/concepts/primitives/attribute.mdx"
locale: "zh-Hans"
---

# Attribute

**Attribute** 是属于一个 Flow run 的 durable 数据。Step 和 RPC 读写它。Dex 把它和 Flow 存在一起，所以中间状态不需要另外的 database。

Attribute 的生命周期和它的 Flow 相同。Flow 关闭后过了 retention period，Dex 删除 Flow，也一并删除 Attribute。要更长时间保留数据，见[在 Flow 之后保留数据](https://docs.superdurable.io/zh-Hans/concepts/primitives/attribute.md#attribute-store-projection)。

## 什么时候用 Attribute

Attribute 适合存放 Flow 据以做决定的状态：订单状态、计数器、客户资料，或者前一个 Step 的结果。Step 要等待并消费的消息，用 [Channel](https://docs.superdurable.io/zh-Hans/concepts/primitives/channel.md)。可以丢失的实时进度，用 [Stream](https://docs.superdurable.io/zh-Hans/concepts/primitives/stream.md)。

<a id="define-an-attribute"></a>

## 使用 Attribute

在 Flow 的 persistence schema 里声明每个 Attribute。之后 Step 或 RPC 就可以读写它。schema 没有声明的 Attribute，Worker 会拒绝。把 Attribute 设为 null 会删除它。

只有 Step 方法或 RPC 成功时，它的 Attribute 写入才会生效。见 [Step 的保证](https://docs.superdurable.io/zh-Hans/concepts/step.md#guarantees)。

**Python**

```python
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](https://github.com/superdurable/dex/tree/main/examples/python/dex_examples/primitives/attribute/attribute_flow.py)

**Go**

```go
var Status = dex.DefineAttribute[string](
	"primitive-attribute-status",
	dex.Indexed(dex.AttributeIndex{Type: dex.IndexKeyword, IndexKey: "keyword1"}),
)

type AttributeFlow struct {
	dex.FlowDefaults
}

func (*AttributeFlow) GetPersistenceSchema() dex.PersistenceSchema {
	return dex.PersistenceSchema{Attributes: []dex.AttributeDef{Status}}
}

type attributeStep struct {
	dex.StepDefaults
}

func (attributeStep) WaitFor(ctx dex.Context, input string) (*dex.Wait, error) {
	if err := Status.Set(ctx, "processing"); err != nil {
		return nil, err
	}
	return dex.SkipWaitImmediately(), nil
}

func (attributeStep) Execute(ctx dex.Context, input string) (*dex.StepDecision, error) {
	if err := Status.Set(ctx, "completed"); err != nil {
		return nil, err
	}
	return dex.GracefulComplete(input), nil
}
```

例子: [examples/go/primitives/attribute/workflow.go](https://github.com/superdurable/dex/tree/main/examples/go/primitives/attribute/workflow.go)

**Java**

```java
public final class AttributeFlow implements Flow<String> {
    private final Attribute<String> status =
            Attribute.define(
                    "primitive-attribute-status",
                    String.class,
                    new AttributeIndex(AttributeIndex.Type.KEYWORD, "keyword1"));
    private final AttributeStep start = new AttributeStep();

    @Override
    public StepList<String> getSteps() {
        return StepList.startStep(start);
    }

    @Override
    public PersistenceSchema getPersistenceSchema() {
        return PersistenceSchema.of(status);
    }

    final class AttributeStep implements Step<String> {
        @Override
        public Wait waitFor(final Context context, final String input) {
            status.set(context, "processing");
            return Wait.skipImmediately();
        }

        @Override
        public StepDecision execute(final Context context, final String input) {
            status.set(context, "completed");
            return StepDecision.gracefulComplete(input);
        }
    }
}
```

例子: [examples/java/src/main/java/io/superdurable/dex/primitives/attribute/AttributeFlow.java](https://github.com/superdurable/dex/tree/main/examples/java/src/main/java/io/superdurable/dex/primitives/attribute/AttributeFlow.java)

**TypeScript**

```typescript
class AttributeStep implements Step<string> {
  public readonly inputCodec = stringCodec;

  public constructor(private readonly flow: AttributeFlow) {}

  public getStepType(): string {
    return "AttributeStep";
  }

  public waitFor(context: Context, input: string): Wait {
    this.flow.status.set(context, "processing");
    return Wait.skipImmediately();
  }

  public execute(context: Context, input: string): StepDecision {
    this.flow.status.set(context, "completed");
    return gracefulComplete(input);
  }
}

export class AttributeFlow implements Flow<string> {
  public readonly status = new Attribute("primitive-attribute-status", stringCodec, {
    type: IndexType.KEYWORD,
    indexKey: "keyword1",
  });
  private readonly start = new AttributeStep(this);

  public getSteps() {
    return StepList.startStep(this.start);
  }

  public getPersistenceSchema(): PersistenceSchema {
    return { attributes: [this.status] };
  }
}
```

例子: [examples/typescript/src/primitives/attribute/attribute-flow.ts](https://github.com/superdurable/dex/tree/main/examples/typescript/src/primitives/attribute/attribute-flow.ts)

**Rust**

```rust
pub struct AttributeFlow {
    status: Attribute<String>,
    email: Attribute<String>,
    progress: AttributeMap<String>,
    start: AttributeStep,
}

impl Default for AttributeFlow {
    fn default() -> Self {
        let status = Attribute::new("primitive-attribute-status")
            .indexed(AttributeIndex::keyword().with_key("keyword1"));
        let email = Attribute::new("primitive-attribute-email").sync_to_attribute_store();
        let progress = AttributeMap::new("primitive-attribute-progress")
            .indexed(AttributeIndex::keyword().with_key("keyword2"));
        Self {
            start: AttributeStep {
                status: status.clone(),
                progress: progress.clone(),
            },
            status,
            email,
            progress,
        }
    }
}

impl Flow for AttributeFlow {
    type StartInput = String;

    fn steps(&self) -> StepList<'_, Self::StartInput> {
        StepList::start(&self.start)
    }

    fn persistence(&self) -> PersistenceSchema {
        PersistenceSchema::new()
            .attribute(&self.status)
            .attribute(&self.email)
            .attribute_map(&self.progress)
    }
}

struct AttributeStep {
    status: Attribute<String>,
    progress: AttributeMap<String>,
}

impl Step for AttributeStep {
    type Input = String;

    fn wait_for(&self, context: &mut Context, _input: Self::Input) -> HandlerResult<Wait> {
        self.status.set(context, "processing".to_owned())?;
        self.progress
            .set(context, "payment", "authorized".to_owned())?;
        Ok(Wait::skip_immediately())
    }

    fn execute(&self, context: &mut Context, input: Self::Input) -> HandlerResult<StepDecision> {
        self.status.set(context, "completed".to_owned())?;
        Ok(StepDecision::graceful_complete(input))
    }
}
```

例子: [examples/rust/src/primitives/attribute/flow.rs](https://github.com/superdurable/dex/tree/main/examples/rust/src/primitives/attribute/flow.rs)

<a id="attributemap"></a>

## AttributeMap

**AttributeMap** 是一组 key 在运行时才确定的 Attribute。它适合随数据增长的值，例如订单里每笔 payment 的状态。Dex 分别存储每个 key，所以一次写入不会重写整个集合。

**Python**

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

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

例子: [examples/python/dex_examples/primitives/attribute/attribute_flow.py](https://github.com/superdurable/dex/tree/main/examples/python/dex_examples/primitives/attribute/attribute_flow.py)

**Go**

```go
var Progress = dex.DefineAttributeMap[string]("primitive-attribute-progress")

if err := Progress.Set(ctx, "payment", "authorized"); err != nil {
	return nil, err
}
```

例子: [examples/go/primitives/attribute/workflow.go](https://github.com/superdurable/dex/tree/main/examples/go/primitives/attribute/workflow.go)

**Java**

```java
private final AttributeMap<String> progress =
        AttributeMap.define("primitive-attribute-progress", String.class);

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

例子: [examples/java/src/main/java/io/superdurable/dex/primitives/attribute/AttributeFlow.java](https://github.com/superdurable/dex/tree/main/examples/java/src/main/java/io/superdurable/dex/primitives/attribute/AttributeFlow.java)

**TypeScript**

```typescript
public readonly progress = new AttributeMap("primitive-attribute-progress", stringCodec);

this.flow.progress.set(context, "payment", "authorized");
```

例子: [examples/typescript/src/primitives/attribute/attribute-flow.ts](https://github.com/superdurable/dex/tree/main/examples/typescript/src/primitives/attribute/attribute-flow.ts)

**Rust**

```rust
let progress = AttributeMap::new("primitive-attribute-progress");

self.progress
    .set(context, "payment", "authorized".to_owned())?;
```

例子: [examples/rust/src/primitives/attribute/flow.rs](https://github.com/superdurable/dex/tree/main/examples/rust/src/primitives/attribute/flow.rs)

<a id="initial-attribute-values"></a>

<a id="read-from-outside-a-flow"></a>

<a id="indexed-attributes"></a>

## 读取和搜索 Attribute

Flow 外的代码通过 [Client](https://docs.superdurable.io/zh-Hans/concepts/client.md) 读取 Attribute。什么时候改用 RPC，见[什么时候读取需要 RPC](https://docs.superdurable.io/zh-Hans/concepts/client.md#related-primitives)。另见 [Read Attributes from outside a Flow](https://docs.superdurable.io/zh-Hans/reference/server/runtime-data.md#read-attributes-from-outside-a-flow)。在 Flow 启动时设置初始值，见 [Initial values](https://docs.superdurable.io/zh-Hans/reference/server/runtime-data.md#initial-attribute-values)。

带 index 的 Attribute 可以用在 **SearchFlows** 查询里，查找 Flow run。上面的例子给 status Attribute 建了 index。index 类型、index key 和 AttributeMap index 见 [Indexes](https://docs.superdurable.io/zh-Hans/reference/server/runtime-data.md#attribute-indexes)。

<a id="locking"></a>

## Lock

两个 Step 或 RPC 可能同时写同一个 Attribute。没有 lock 时，两个先读后写这个 Attribute 的 Step 或 RPC 可能互相覆盖对方的更新，见 [Step 的保证](https://docs.superdurable.io/zh-Hans/concepts/step.md#guarantees)。用 **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：

**Python**

```python
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](https://github.com/superdurable/dex/tree/main/examples/python/dex_examples/primitives/attribute/attribute_flow.py)

**Go**

```go
func (attributeStep) GetStepOptions() *dex.StepOptions {
	return &dex.StepOptions{
		ExecuteLockAttributes: []dex.AttributeLock{
			dex.LockAttribute(Status),
			dex.LockAttributeMap(Progress, "payment"),
		},
	}
}
```

例子: [examples/go/primitives/attribute/workflow.go](https://github.com/superdurable/dex/tree/main/examples/go/primitives/attribute/workflow.go)

**Java**

```java
return StepOptions.newBuilder()
        .addExecuteLock(AttributeLock.of(status))
        .addExecuteLock(AttributeLock.of(progress, "payment"))
        .build();
```

例子: [examples/java/src/main/java/io/superdurable/dex/primitives/attribute/AttributeFlow.java](https://github.com/superdurable/dex/tree/main/examples/java/src/main/java/io/superdurable/dex/primitives/attribute/AttributeFlow.java)

**TypeScript**

```typescript
return {
  executeLockAttributes: [this.flow.status.lock(), this.flow.progress.lock("payment")],
};
```

例子: [examples/typescript/src/primitives/attribute/attribute-flow.ts](https://github.com/superdurable/dex/tree/main/examples/typescript/src/primitives/attribute/attribute-flow.ts)

**Rust**

```rust
StepOptions::new()
    .execute_lock(self.status.lock())
    .execute_lock(self.progress.lock("payment"))
```

例子: [examples/rust/src/primitives/attribute/flow.rs](https://github.com/superdurable/dex/tree/main/examples/rust/src/primitives/attribute/flow.rs)

<a id="whole-map-coordination"></a>

<a id="lock-waitfor"></a>

<a id="lock-execute"></a>

<a id="lock-an-rpc"></a>

<a id="rpc-lock-conflicts"></a>

lock 的获取时机，以及 **WaitFor**、**Execute** 和 RPC 的 lock，见 [Locks](https://docs.superdurable.io/zh-Hans/reference/server/runtime-data.md#attribute-locks)。lock conflict 见 [RPC lock conflicts](https://docs.superdurable.io/zh-Hans/reference/server/runtime-data.md#rpc-lock-conflicts)，锁住整张 AttributeMap 见 [whole-map coordination](https://docs.superdurable.io/zh-Hans/reference/server/runtime-data.md#whole-map-coordination)。

<a id="hard-delete-attributes"></a>

## 短期 secret

**hard-delete Attribute** 用来保存短期 secret，例如 Flow 用几次就丢弃的 API key。删除它会永久清除存储的值。Dex API 和 Flow history 都无法恢复。和其他写入不同，hard-delete 写入在 Dex 保存 handler 的结果之前就生效。要求和后果见 [Hard-delete Attributes](https://docs.superdurable.io/zh-Hans/reference/server/runtime-data.md#hard-delete-attributes)。

<a id="与其他组件的关系"></a>

<a id="配置"></a>

<a id="交付与重试"></a>

<a id="attribute-store-local-development"></a>

<a id="attribute-store-projection"></a>

## 在 Flow 之后保留数据

Attribute Store sync 把选中的 Attribute 复制到你自己的 PostgreSQL 表。每个目标称为一个 **Attribute Store**。Dex 删除 Flow 后，表里的数据仍然保留。你的应用和报表用 SQL 查询这些表，你也可以为自己的查询建 index。

同步一个 Attribute 的步骤：

1. 在 server 启动前建好表。Dex 不会创建、迁移或回填表。
2. 在 server 配置里配置这个 store。
3. 用 **syncToAttributeStore** 标记 Attribute 或 AttributeMap。
4. 启动 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](https://docs.superdurable.io/zh-Hans/reference/server/attribute-store-sync.md) 参考页。完整示例见 [Entity Store](https://docs.superdurable.io/zh-Hans/design-patterns/entity-store.md) 模式。

## 参考

每个 SDK 参考的 [Attribute](https://docs.superdurable.io/zh-Hans/reference/python/attribute.md) 页列出该语言的 API 名称。

[Data primitive semantics](https://docs.superdurable.io/zh-Hans/reference/server/runtime-data.md#attributes) 详细说明初始值、Client 读取、index、hard-delete Attribute 和 lock。