Skip to main content

persistence

PersistenceBackend​

from dara.core.persistence import PersistenceBackend

class PersistenceBackend(BaseModel, abc.ABC)

Abstract base class for a BackendStore backend.

Methods​

has​

@abc.abstractmethod
async def has(key: str) -> bool

Check if a key exists

write​

@abc.abstractmethod
async def write(key: str, value: Any)

Persist a value

read​

@abc.abstractmethod
async def read(key: str) -> Any

Read a value

delete​

@abc.abstractmethod
async def delete(key: str)

Delete a value

get_all​

@abc.abstractmethod
async def get_all() -> Dict[str, Any]

Get all the values as a dictionary of key-value pairs

subscribe​

async def subscribe(on_value: Callable[[str, Any], Awaitable[None]])

Subscribe to changes in the backend. Called with a callback that should be invoked whenever a value is updated.

InMemoryBackend​

from dara.core.persistence import InMemoryBackend

class InMemoryBackend(PersistenceBackend)

In-memory persistence backend

Attributes​

  • data: Dict[str, Any]

FileBackend​

from dara.core.persistence import FileBackend

class FileBackend(PersistenceBackend)

File persistence backend implementation.

Stores data in a JSON file

Attributes​

  • path: str
  • _lock: aiorwlock.RWLock

PersistenceStore​

from dara.core.persistence import PersistenceStore

class PersistenceStore(BaseModel, abc.ABC)

Base class for a variable persistence store

Methods​

init​

@abc.abstractmethod
async def init(variable: 'Variable')

Initialize the store when connecting to a variable

BackendStore​

from dara.core.persistence import BackendStore

class BackendStore(PersistenceStore)

Persistence store implementation that uses a backend implementation to store data server-side

Arguments:

  • uid: unique identifier for this store; defaults to a random UUID
  • backend: the backend to use for storing data; defaults to an in-memory backend
  • scope: the scope for the store; if 'global' a single value is stored for all users, if 'user' a value is stored per user
  • readonly: whether to use the backend in read-only mode, i.e. skip syncing values from client to backend and raise if write()/delete() is called

Attributes​

  • uid: str
  • backend: PersistenceBackend
  • scope: Literal[]
  • readonly: bool
  • default_value: Any
  • initialized_scopes: Set[str]
  • sequence_number: Dict[str, int]

Methods​

__init__​

def __init__(backend: Optional[PersistenceBackend] = None,
             uid: Optional[str] = None,
             scope: Optional[str] = None,
             readonly: bool = False)

Persistence store implementation that uses a backend implementation to store data server-side

Arguments:

  • uid: unique identifier for this store; defaults to a random UUID
  • backend: the backend to use for storing data; defaults to an in-memory backend
  • scope: the scope for the store; if 'global' a single value is stored for all users, if 'user' a value is stored per user
  • readonly: whether to use the backend in read-only mode, i.e. skip syncing values from client to backend and raise if write()/delete() is called

init​

async def init(variable: 'Variable')

Write the default value to the store if it's not set

Arguments:

  • variable: the variable to initialize the store for

write_partial​

async def write_partial(data: Union[List[Dict[str, Any]], Any],
                        notify: bool = True)

Apply partial updates to the store using JSON Patch operations or automatic diffing.

If scope='user', the patches are applied for the current user so the method can only be used in authenticated contexts.

Arguments:

  • data: Either a list of JSON patch operations (RFC 6902) or a full object to diff against current value
  • notify: whether to broadcast the patches to clients

write​

async def write(value: Any, notify=True, ignore_channel: Optional[str] = None)

Persist a value to the store.

If scope='user', the value is written for the current user so the method can only be used in authenticated contexts.

Arguments:

  • value: value to write
  • notify: whether to broadcast the new value to clients
  • ignore_channel: if passed, ignore the specified websocket channel when broadcasting

read​

async def read()

Read a value from the store.

If scope='user', the value is read for the current user so the method can only be used in authenticated contexts.

delete​

async def delete(notify=True)

Delete the persisted value from the store

If scope='user', the value is deleted for the current user so the method can only be used in authenticated contexts.

Arguments:

  • notify: whether to broadcast that the value was deleted to clients

get_all​

async def get_all() -> Dict[str, Any]

Get all the values from the store as a dictionary of key-value pairs.

For global scope, the dictionary contains a single key-value pair {'global': value}. For user scope, the dictionary contains a key-value pair for each user {'user1': value1, 'user2': value2, ...}.

BackendStoreEntry​

from dara.core.persistence import BackendStoreEntry

class BackendStoreEntry(BaseModel)

Attributes​

  • uid: str
  • store: BackendStore

store​

Store instance