StateStore

StateStore provides persistent state management for workflows and credentials in the Atlan Apps Framework. It uses Dapr's state management component to store and retrieve application state, supporting both single-field updates and complete state object operations.

StateStore component

StateStore is defined as a Dapr component through YAML configuration. When Dapr starts, it loads all component YAML files from the /components folder, detects statestore.yaml, and initializes a StateStore component of type state.sqlite. The application can then use save, get, or delete methods of DaprClient to perform operations on the state.

Reference: statestore.yaml

Component definition

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: statestore
spec:
  version: v1
  type: state.sqlite
  ignoreErrors: true
  metadata:
    # Connection string
    - name: connectionString
      value: "./local/dapr/statestore.db"

Definition: statestore.py

StateType enum

StateStore can store state related to workflows or credentials. The state type is controlled through the StateType enum.

Reference: StateType enum

class StateType(Enum):
    WORKFLOWS = "workflows"
    CREDENTIALS = "credentials"

The enum includes an is_member class method that checks if a value is a valid state type.

`StateType`Enum
Optional

Enumeration of state types supported by StateStore. Use WORKFLOWS for workflow execution state, arguments, and configuration. Use CREDENTIALS for credential configurations and secrets.

Allowed values:

WORKFLOWSCREDENTIALS

Examples:

StateType.WORKFLOWS
StateType.CREDENTIALS

Example:

>>> StateType.is_member("workflows")
True

build_state_store_path

Builds a unique path for state storage based on the identifier and state type.

`id`string
Required

Unique identifier for the state. Used to construct the storage path along with the state type.

Examples:

"workflow-123"
"db-cred-456"
"my-workflow-id"

`state_type`StateType
Required

Type of state (WORKFLOWS or CREDENTIALS). Determines the subdirectory path where the state is stored.

Examples:

StateType.WORKFLOWS
StateType.CREDENTIALS

Returns

`path`string
Optional

Unique file path for the state. The path format is './local/tmp/persistent-artifacts/apps/{appName}/{state_type}/{id}/config.json'.

Example:"./local/tmp/persistent-artifacts/apps/appName/workflows/workflow-123/config.json"

Example

from application_sdk.services.statestore import build_state_store_path, StateType

# Workflow state path
path = build_state_store_path("workflow-123", StateType.WORKFLOWS)
print(path)

Output: './local/tmp/persistent-artifacts/apps/appName/workflows/workflow-123/config.json'

The unique identifier and path for state depends on its identifier plus the type of state. This path can be used in get_state and set_state functions.

StateStore class

The StateStore class provides methods for managing application state. All methods are class methods that operate on the state store.

The StateStore class includes the following methods:

get_state - Retrieves state from the store based on identifier and type
save_state - Saves or updates a single field in the state store
save_state_object - Saves an entire state object to the object store

get_state

Retrieves state from the store based on identifier and type.

`id`string
Required

Unique identifier for the state. Must match the identifier used when saving the state.

Examples:

"workflow-123"
"db-cred-456"
"my-workflow-id"

`type`StateType
Required

Type of state (WORKFLOWS or CREDENTIALS). Must match the type used when saving the state.

Examples:

StateType.WORKFLOWS
StateType.CREDENTIALS

Returns

`state`dict | None
Optional

State dictionary if present, otherwise None. The dictionary contains the stored state fields as key-value pairs. Returns None if no state exists for the given identifier and type.

Examples:

{"status": "running", "progress": 50}
{"database": "mydb", "host": "localhost"}
None

Examples

from application_sdk.services.statestore import StateStore, StateType

# Get workflow state
state = await StateStore.get_state("workflow-123", StateType.WORKFLOWS)
print(f"Current status: {state.get('status', 'unknown')}")

# Get credential configuration
creds = await StateStore.get_state("db-cred-456", StateType.CREDENTIALS)
print(f"Database: {creds.get('database')}")

This method internally retrieves state from the object store.

save_state

Saves or updates a single field in the state store.

`key`string
Required

Field name to save or update. This becomes a key in the state dictionary.

Examples:

"progress"
"status"
"current_step"
"config"

`value`any
Required

Value to store. Can be any JSON-serializable type (string, number, dict, list, bool, etc.).

Examples:

75
"running"
{"batch_size": 1000}
[1, 2, 3]
true

`id`string
Required

Unique identifier for the state. Used to locate the state object in the store.

Examples:

"workflow-123"
"db-cred-456"
"my-workflow-id"

`type`StateType
Required

Type of state (WORKFLOWS or CREDENTIALS). Determines the storage location and namespace.

Examples:

StateType.WORKFLOWS
StateType.CREDENTIALS

Returns

`None`None
Optional

This method returns None. The state is saved asynchronously to the object store.

Example

from application_sdk.services.statestore import StateStore, StateType

# Update workflow progress
await StateStore.save_state(
    key="progress",
    value=75,
    id="workflow-123",
    type=StateType.WORKFLOWS
)

This method internally saves state to a local path by creating appropriate directories and files, then uploads to the object store. Use this method only for creating or updating a single field.

save_state_object

Saves an entire state object to the object store.

`id`string
Required

Unique identifier for the state. Used to locate the state object in the store.

Examples:

"workflow-123"
"db-cred-456"
"my-workflow-id"

`value`dict
Required

Complete state object to save. Replaces the entire existing state object with this dictionary. All fields must be JSON-serializable (string, number, dict, list, bool, None).

Examples:

{"status": "running", "progress": 50, "config": {"batch_size": 1000}}
{"database": "mydb", "host": "localhost", "port": 5432}
{"workflow_id": "wf-123", "workflow_run_id": "run-456", "output_path": "/data/output"}

`type`StateType
Required

Type of state (WORKFLOWS or CREDENTIALS). Determines the storage location and namespace.

Examples:

StateType.WORKFLOWS
StateType.CREDENTIALS

Returns

`updated`dict
Optional

The saved state object. Returns the complete dictionary that was saved to the store, confirming the operation succeeded.

Examples:

{"status": "running", "progress": 50, "config": {"batch_size": 1000}}
{"workflow_id": "wf-123", "workflow_run_id": "run-456"}

Example

from application_sdk.services.statestore import StateStore, StateType

# Save complete workflow state
workflow_state = {
    "status": "running",
    "current_step": "data_processing",
    "progress": 50,
    "config": {"batch_size": 1000}
}
updated = await StateStore.save_state_object(
    id="workflow-123",
    value=workflow_state,
    type=StateType.WORKFLOWS
)
print(f"Final state has {len(updated)} keys")

Use this method for creating a multi-field state or updating multiple fields in a state.

Usage patterns

Workflow arguments

Workflow arguments are stored in state with workflow_id as the key. They contain information such as:

output_prefix
output_path
workflow_id
workflow_run_id

The get_workflow_args activity uses the get_state method from StateStore to fetch these arguments.

TemporalClient integration

When a new workflow is started through TemporalWorkflowClient, the start_workflow method adds application_name and workflow_id to the workflow arguments and saves the entire object to the state store using the save_state_object method.

SecretStore integration

SecretStore uses internal utility functions like _get_credentials_async and save_secret to save to StateStore or fetch from the state store using credential_guid as the key.

Tests

Tests for the StateStore service are defined in the test_statestore.py file. Both success and failure scenarios are tested, including errors ranging from JSONDecodeError to object store upload failures.

StateStore component​

Component definition​

StateType enum​

StateTypeEnumOptional

build_state_store_path​

idstringRequired

state_typeStateTypeRequired

Returns​

pathstringOptional

Example​

StateStore class​

get_state​

idstringRequired

typeStateTypeRequired

Returns​

statedict | NoneOptional

Examples​

save_state​

keystringRequired

valueanyRequired

idstringRequired

typeStateTypeRequired

Returns​

NoneNoneOptional

Example​

save_state_object​

idstringRequired

valuedictRequired

typeStateTypeRequired

Returns​

updateddictOptional

Example​

Usage patterns​

Workflow arguments​

TemporalClient integration​

SecretStore integration​

Tests​

See also​

StateStore component

Component definition

StateType enum

`StateType`Enum
Optional

build_state_store_path

`id`string
Required

`state_type`StateType
Required

Returns

`path`string
Optional

Example

StateStore class

get_state

`id`string
Required

`type`StateType
Required

Returns

`state`dict | None
Optional

Examples

save_state

`key`string
Required

`value`any
Required

`id`string
Required

`type`StateType
Required

Returns

`None`None
Optional

Example

save_state_object

`id`string
Required

`value`dict
Required

`type`StateType
Required

Returns

`updated`dict
Optional

Example

Usage patterns

Workflow arguments

TemporalClient integration

SecretStore integration

Tests

See also