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, and queryable 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] = {}

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

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.