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

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 read and write Attributes after a Flow starts. 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.

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 type	Value
KEYWORD	One exact string.
FULL_TEXT	Text for tokenized search.
KEYWORD_ARRAY	A list of exact strings.
INT, DOUBLE, BOOL	The corresponding scalar value.
DATETIME	A date and time value.

An index key is the key in the search schema and the key you use in a query. Index keys are shared across Flow types.

A regular Attribute uses its Attribute name as the index key by default. You can set an explicit index key instead.

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

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

For example, the status Attribute above uses OrderStatus as its index key. To find Flow executions whose status is completed, use this query:

OrderStatus = 'completed'

Pass the query to SearchFlows:

page = await app_state.client.search_flows(
    "OrderStatus = '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:

OrderProgress = 'authorized'

If another Map instance later writes a different value, that value replaces the search value for OrderProgress. 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, "OrderProgress"),
)

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

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 Dex Server configuration:

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

Start Dex Server with the configuration:

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.

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