Skip to main content

Data Storage

Keep one durable Flow for each application entity, then project selected Attributes into an application-owned SQL table. The Flow is the write model and source of truth. SQL is an asynchronous, latest-state projection for reads and queries.

The runnable example keeps one UserProfileFlow per user. Its Flow ID becomes the user_id primary key in public.user_profiles. The Server configuration names the projection target entityStore.

When to use this pattern​

Use this pattern when an entity needs durable state or RPC updates, while the application also needs database queries over its latest state. Do not use the SQL projection to make Flow decisions: Dex state remains authoritative.

How it works​

  1. Use the entity ID as the Flow ID.
  2. Mark every projected Attribute with syncToAttributeStore.
  3. Select the Server-configured Attribute Store in FlowConfig when starting the Flow.
  4. Update or delete Attributes through the Flow. Dex projects their resulting values to SQL.

The example projects text, boolean, integer, double, datetime, and JSON values. Deleting an Attribute keeps the SQL row and projects NULL to its column.

Core implementation​

Define synced Attributes​

Every SDK marks the same profile fields for projection. Their Flow persistence schemas include the definitions, so the values are durable before Dex syncs them.

class UserProfileFlow(Flow[None]):
    display_name = Attribute("display_name", str, sync_to_attribute_store=True)
    email = Attribute("email", str, sync_to_attribute_store=True)
    marketing_opt_in = Attribute(
        "marketing_opt_in", bool, sync_to_attribute_store=True
    )
    credits = Attribute("credits", int, sync_to_attribute_store=True)
    weight = Attribute("weight", float, sync_to_attribute_store=True)
    last_logged_in_time = Attribute(
        "last_logged_in_time", datetime, sync_to_attribute_store=True
    )
    metadata = Attribute(
        "metadata", UserProfileMetadata, sync_to_attribute_store=True
    )

    def get_persistence_schema(self) -> PersistenceSchema:
        return PersistenceSchema.of(
            self.display_name,
            self.email,
            self.marketing_opt_in,
            self.credits,
            self.weight,
            self.last_logged_in_time,
            self.metadata,
        )

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

Update through a Flow RPC​

The RPC is the write path. Each implementation validates the profile at its boundary, then updates every projected Attribute.

@rpc
def update_profile(self, context: Context, input: UserProfile) -> None:
    input.validate()
    self.display_name.set(context, input.display_name)
    self.email.set(context, input.email)
    self.marketing_opt_in.set(context, input.marketing_opt_in)
    self.credits.set(context, input.credits)
    self.weight.set(context, input.weight)
    self.last_logged_in_time.set(context, input.last_logged_in_time)
    self.metadata.set(context, input.metadata)

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

Select the storage target at Flow start​

The Attribute Store is configured on the Server. Select it in FlowConfig when starting the Flow; the initial Attribute values in that same request also project to SQL.

config_override=FlowConfig(attribute_store_names=[STORE_NAME]),

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

API endpoints​

  • POST /patterns/entity-store/profile — create a profile Flow
  • POST /patterns/entity-store/profile/update — update its synced fields
  • GET /patterns/entity-store/profile?userId=... — read authoritative Flow state
  • POST /patterns/entity-store/profile/clear?userId=... — delete the synced values

Run with PostgreSQL​

The shared example setup starts PostgreSQL, creates public.user_profiles, and passes the entityStore configuration to dexcli dev. It includes the table schema and curl commands for every language example.