Skip to main content

Attribute

An Attribute lets you store data within a Flow execution, as persistence storage.

This is especially useful for intermediate data during a Flow execution. It eliminates the work of setting up and managing a database dependency.

Dex Attribute is not to replace your databases. An obvious limitation is that Attribute exists only for the lifetime of its Flow execution. Dex deletes it when it deletes the Flow execution, after the configurable retention period following Flow closure.

However, Dex allows you to sync Attribute values to your application database as permanent storage. This also lets you build more powerful database indexes to query Attributes.

Use an Attribute​

Define Attributes as part of the persistence schema of a Flow.

Then you can use the attribute within Step or RPC implementation to read or write the attribute. An attribute must be declared via PersistenceSchema before use. Otherwise, the worker will reject the operations.

When updating an attribute, setting it to null will delete it.

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)

Example: examples/python/dex_examples/primitives/attribute/attribute_flow.py

AttributeMap​

An AttributeMap is a family of Attributes with keys decided at runtime. Use one when a Flow needs a changing set of small values, such as the status of each payment or each item in an order.

The reason for AttributeMap is to efficiently store a collection of data. Each Attribute is stored as a blob in Dex Server. Putting all data into a single Attribute means every update rewrites the whole blob. AttributeMap lets Dex Server store each instance as an independent blob for higher efficiency.

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

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

Example: examples/python/dex_examples/primitives/attribute/attribute_flow.py

Initial Attribute values​

StartFlowOptions can set initial Attributes when the Flow is created. Initial values and Flow creation are one operation, so the first Step sees them immediately. See Flow options.

Applications use Flow RPCs to write Attributes after a Flow starts. To read them from outside the Flow, see Read Attributes from outside a Flow. A Client can also use WaitForAttributeMatch to wait until a singleton Attribute or one AttributeMap instance satisfies a scalar comparison. The call returns the matched current value. See Client.

Read Attributes from outside a Flow​

Application code outside the Flow reads Attribute values with the Client's GetAttributes call. One call reads any mix of singleton Attributes and AttributeMap instances, and all values come from the same commit. Dex Server answers the read itself, so no RPC handler or Worker runs.

The call reads the current run, or the latest run when no run is open. A closed run stays readable until Dex purges it. The call sees only committed state, while an RPC handler also sees its own earlier writes in the same invocation. Dex Server resolves hard-delete values before it responds, and a deleted value reads as unset, like any Attribute without a stored value.

An RPC remains the right choice when the read needs Flow-side validation, authorization, a derived view, or Attribute locks. See Client: Read Attributes for the example, the unset value in each SDK, and the errors.

Indexed Attributes​

Dex lets you configure indexes for Attributes. With indexing, you can use the field in SearchFlows queries to find the Flow executions that match the query.

Index typeValue
KEYWORDOne exact string.
FULL_TEXTText for tokenized search.
KEYWORD_ARRAYA list of exact strings.
INT, DOUBLE, BOOLThe corresponding scalar value.
DATETIMEA date and time value.

An index key is the key in the search schema and the key you use in a query. Index keys consume a finite, namespace-level Search Attribute pool and are shared across Flow types.

The SDK defaults a regular Attribute's index key to its Attribute name. For application indexes, always override that default with an explicit generic numbered key. Keep business names on Attributes and share one typed slot pool across every Flow type in the namespace. Two logical indexed fields in the same Flow need different slots. Reuse a slot across Flow types only with the same index type; its business meaning can differ. Dex system indexes keep their reserved names.

Index typeApplication index keys
KEYWORDkeyword1, keyword2, …
KEYWORD_ARRAYkeywordList1, keywordList2, …
FULL_TEXTtext1, text2, …
INTint1, int2, …
BOOLbool1, bool2, …
DOUBLEdouble1, double2, …
DATETIMEdatetime1, datetime2, …

Renaming an index does not backfill existing Flow executions. Validate changed example schemas in a fresh local dexcli dev instance. Plan any existing deployment migration separately.

A generic key can have a different business meaning in each Flow type. Application run queries must constrain FlowType. Keep the query unchanged across pages. Dex Web already scopes searches to the selected Flow type. Parenthesize caller filters before combining them with the application scope so an OR cannot bypass it. Do not index sensitive data or PII. The customer-email value in the Customer Refund example is synthetic data only.

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

Example: examples/python/dex_examples/primitives/attribute/attribute_flow.py

For example, the status Attribute above uses keyword1 as its index key. To find completed AttributeFlow executions, use this query:

FlowType = 'AttributeFlow' AND keyword1 = 'completed'

Pass the query to SearchFlows:

page = await app_state.client.search_flows(
    "FlowType = 'AttributeFlow' AND keyword1 = 'completed'",
    20,
    "",
)

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

An indexed AttributeMap must use an explicit index key. It is not an index over all Map instances. Each instance write updates the same search field, so one instance can replace the value written by another instance. The Map instance key is not searchable.

This does not turn the Map into a KEYWORD_ARRAY. KEYWORD_ARRAY indexes the string array from one write; Dex does not combine the values from all Map instances. After the example below writes the payment instance with value authorized, you can query the current search value with:

FlowType = 'AttributeFlow' AND keyword2 = 'authorized'

If another Map instance later writes a different value, that value replaces the search value for keyword2. Use a regular Attribute with KEYWORD_ARRAY when you need to search a collection of values. Do not use an AttributeMap index to find every instance with a value.

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

Example: examples/python/dex_examples/primitives/attribute/attribute_flow.py

Hard-delete Attributes​

Use a hard-delete Attribute for a short-lived secret, such as an API key that a Flow uses a few times and then discards.

Dex stores every value of a hard-delete Attribute only in object storage, whatever the value size. Flow history keeps only a reference. Deleting the Attribute, or setting it to null, permanently deletes the stored value. After that, neither Dex APIs nor Flow history can recover it.

From server-zd v0.1.5, a direct call to the FlowService SetAttributes API can erase a hard-delete Attribute after its Flow run closes and before the run is purged. The SDK Clients do not expose this API, so send it with dexcli api call SetAttributes or another FlowService client. Set every Attribute in the request to null. Each Attribute must still have a stored hard-delete value. The erase leaves the run closed and permanently deletes its stored objects. Other writes to a closed run return NOT_FOUND.

Step and RPC handlers receive the value directly. The SDKs never write it to their local blob cache, and Dex Server never writes it to its blob cache.

Enable it on the definition with HardDeleteOnNull in Go, hardDeleteOnNull in Java and TypeScript, and hard_delete_on_null in Python and Rust. The option works for both Attributes and AttributeMaps.

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)

Example: examples/python/dex_examples/primitives/attribute/attribute_flow.py

A hard-delete Attribute has these requirements:

  • Only string and object values are supported, including bytes. Numeric and Boolean value types are rejected.
  • It cannot be indexed or synced to an Attribute Store. WaitForAttributeMatch and Action conditions reject it.
  • A server that supports hard-delete Attributes, such as server-zd. The SDKs refuse to write the Attribute to an older server.

Each Flow keeps one stored value per Attribute key and overwrites it in place. Plan for these consequences:

  • A Step or RPC attempt writes or deletes the stored value before its result is committed. A retried attempt, other Steps and RPCs, and Attribute reads can observe a write or deletion made by an attempt that later failed. Make handlers safe to repeat instead of relying on a read-modify-write of the Attribute.
  • When concurrent Steps or RPCs write the same key, the final value may not be the last committed write.
  • A Flow start or SubFlow start that fails because the Flow ID is already running still overwrites the running Flow's value. A null initial value deletes it.
  • Flows that reuse a Flow ID share the stored value.
  • A deleted value always reads as absent. Time travel does not restore it. A restarted SubFlow starts from the original initial value, so it reads as absent when an earlier run deleted that value.
  • Removing the option from a definition leaves the stored value in place until Dex removes the Flow. Delete the value with the hard-delete definition first.

A value remains stored until it is deleted, or until Dex removes the Flow after its retention period. Object storage backups and bucket versioning outside Dex can keep copies. See Object storage.

Attribute Sync​

Dex Attribute is not to replace your databases. Attribute Sync feature will sync the attribute values to your application database, without you writing a single line of code.

So that you can store the attributes in your database as permanent storage. This also lets you build more powerful database indexes to query Attributes.

Mark each Attribute that should sync:

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

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

First, define an Attribute Store in a YAML file:

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

Only PostgreSQL is supported as an Attribute Store. Redis support is tracked in issue 616. Create the destination first. Attribute Sync writes latest state asynchronously and does not automatically backfill existing Flows. See Custom Attribute Store operations for target contracts, type mappings, and retry behavior.

Start the local server with the file:

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

Then choose the same store name when starting the Flow:

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

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

Locking​

Concurrent Steps and RPC invocations can read and write the same Attribute. This can lead to race conditions. To avoid this, use Attribute locks. Dex Server ensures only one Step execution or RPC handling can acquire the lock at a time.

Coordinate writes across the whole AttributeMap​

An AttributeMap lock protects one exact instance. When an invariant spans several instances or the whole collection, register a separate singleton bool Attribute as its coordination key. For example, register currentMessages as an AttributeMap and currentMessagesLock as a non-map, non-indexed bool Attribute in the same PersistenceSchema.

Every Step, timeout handler, and RPC that writes, deletes, archives, or otherwise changes the protected map must request the same singleton Attribute lock in its method-specific lock options. RPCs declare this fixed lock at registration time. A writer that requests only a map-instance lock does not conflict with the singleton lock. All participating writers serialize within that Flow execution; other Flow executions have their own locks.

The bool value is not the lock state. Do not acquire or release the lock by reading or toggling the value. Dex acquires and releases it around the invocation; the coordination Attribute does not need a persisted value. Locking also does not load map entries. Declare exact instance loads, or a whole-map load when the operation must enumerate instances. RPC lock acquisition can return a lock-conflict error rather than queueing the caller.

Keep per-instance locks for independent partition updates. Use the shared singleton lock when all affected writers must preserve a collection-wide invariant.

Lock WaitFor​

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

Example: 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"),
        ),
    )

Example: 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)

Example: examples/python/dex_examples/primitives/attribute/attribute_flow.py