State Stores

Abstract interface and implementations for persisting entity state.

StateStoreClient

StateStoreClient(connection_string: str | None = None)

Base class for persistent state store implementations.

Provides connect/disconnect lifecycle, context manager support, and connection guards. Mirrors pycharter's MetadataStoreClient.

Subclasses must override connect() and should call _require_connection() at the top of every method that touches the backend.

Source code in src/pystator/stores/base.py

def __init__(self, connection_string: str | None = None) -> None:
    self.connection_string = connection_string
    self._connection: Any = None

connect

connect() -> None

Establish backend connection. Subclasses must implement.

Source code in src/pystator/stores/base.py

def connect(self) -> None:
    """Establish backend connection.  Subclasses must implement."""
    raise NotImplementedError("Subclasses must implement connect()")

disconnect

disconnect() -> None

Close the backend connection.

Source code in src/pystator/stores/base.py

def disconnect(self) -> None:
    """Close the backend connection."""
    if self._connection is not None:
        self._connection = None

InMemoryStateStore

InMemoryStateStore()

In-memory state store for testing and simple use cases.

Implements sync, async, versioned, batch, queryable, and event store protocols. Not suitable for production.

Source code in src/pystator/stores/base.py

def __init__(self) -> None:
    self._state: dict[str, str] = {}
    self._context: dict[str, dict[str, Any]] = {}
    self._metadata: dict[str, dict[str, Any]] = {}
    self._versions: dict[str, int] = {}
    self._event_log: dict[str, list[dict[str, Any]]] = {}
    self._event_sequences: dict[str, int] = {}

get_workflow_status

get_workflow_status(entity_id: str) -> str | None

Return persisted workflow_status (active / complete / failed), if any.

Source code in src/pystator/stores/base.py

def get_workflow_status(self, entity_id: str) -> str | None:
    """Return persisted ``workflow_status`` (``active`` / ``complete`` / ``failed``), if any."""
    meta = self._metadata.get(entity_id)
    if meta is None:
        return None
    val = meta.get("workflow_status")
    return val if isinstance(val, str) else None

get_states_batch

get_states_batch(entity_ids: list[str]) -> dict[str, str | None]

Fetch multiple entity states at once.

Source code in src/pystator/stores/base.py

def get_states_batch(self, entity_ids: list[str]) -> dict[str, str | None]:
    """Fetch multiple entity states at once."""
    return {eid: self._state.get(eid) for eid in entity_ids}

set_states_batch

set_states_batch(updates: dict[str, str], *, metadata: dict[str, Any] | None = None) -> None

Set multiple entity states at once.

Source code in src/pystator/stores/base.py

def set_states_batch(
    self,
    updates: dict[str, str],
    *,
    metadata: dict[str, Any] | None = None,
) -> None:
    """Set multiple entity states at once."""
    for eid, state in updates.items():
        self.set_state(eid, state, metadata=metadata)

list_entities

list_entities(state: str | None = None, limit: int = 100, offset: int = 0) -> list[str]

List entity IDs, optionally filtered by state.

Source code in src/pystator/stores/base.py

def list_entities(
    self,
    state: str | None = None,
    limit: int = 100,
    offset: int = 0,
) -> list[str]:
    """List entity IDs, optionally filtered by state."""
    if state is not None:
        ids = [eid for eid, s in self._state.items() if s == state]
    else:
        ids = list(self._state.keys())
    return ids[offset : offset + limit]

count_by_state

count_by_state() -> dict[str, int]

Return count of entities per state.

Source code in src/pystator/stores/base.py

def count_by_state(self) -> dict[str, int]:
    """Return count of entities per state."""
    counts: dict[str, int] = {}
    for s in self._state.values():
        counts[s] = counts.get(s, 0) + 1
    return counts

append_event

append_event(entity_id: str, trigger: str, *, payload: dict[str, Any] | None = None, event_id: str | None = None, source: str = '', machine_name: str | None = None) -> int

Append an event and return its sequence number.

Source code in src/pystator/stores/base.py

def append_event(
    self,
    entity_id: str,
    trigger: str,
    *,
    payload: dict[str, Any] | None = None,
    event_id: str | None = None,
    source: str = "",
    machine_name: str | None = None,
) -> int:
    """Append an event and return its sequence number."""
    seq = self._event_sequences.get(entity_id, 0) + 1
    self._event_sequences[entity_id] = seq
    record = {
        "entity_id": entity_id,
        "sequence": seq,
        "trigger": trigger,
        "payload": dict(payload) if payload else {},
        "event_id": event_id,
        "source": source,
        "machine_name": machine_name,
    }
    self._event_log.setdefault(entity_id, []).append(record)
    return seq

get_events

get_events(entity_id: str, *, after_sequence: int = 0, limit: int = 1000) -> list[dict[str, Any]]

Return events for entity_id ordered by sequence.

Source code in src/pystator/stores/base.py

def get_events(
    self,
    entity_id: str,
    *,
    after_sequence: int = 0,
    limit: int = 1000,
) -> list[dict[str, Any]]:
    """Return events for *entity_id* ordered by sequence."""
    events = self._event_log.get(entity_id, [])
    filtered = [e for e in events if e["sequence"] > after_sequence]
    return filtered[:limit]

Other backends (SQLiteStateStore, PostgresStateStore, MongoDBStateStore, RedisStateStore) are available from pystator.stores when the corresponding optional dependencies are installed. See the package structure and state stores guide.