Storage

Storage is where persisted widget state is saved. A dashboard has none until you give it one, see Persistent Storage.

import dachshund
from dachshund.storage import SQLiteStorage

dachshund.init("Ada", storage=SQLiteStorage("widget_state.db"))

dachshund.storage.protocol.Storage

Bases: Protocol

Where persisted widget state is saved.

A backend maps a key to a JSON-shaped dict, through two async methods: save(key, value) and load(key), which returns None for an unknown key. See the Persistent Storage guide.

Example

import dachshund

dachshund.init("Ada", storage=MyStorage())

save async

save(key, value)

Replace whatever was kept under a key.

Source code in dachshund/storage/protocol.py

async def save(self, key: str, value: dict[str, Any]) -> None:
    """Replace whatever was kept under a key."""
    ...

load async

load(key)

What was last saved under a key, or None if nothing was.

Source code in dachshund/storage/protocol.py

async def load(self, key: str) -> dict[str, Any] | None:
    """What was last saved under a key, or `None` if nothing was."""
    ...

dachshund.storage.sqlite.SQLiteStorage

SQLiteStorage(path)

A Storage in a SQLite file of its own.

Example

import dachshund
from dachshund.storage import SQLiteStorage

dachshund.init("Ada", storage=SQLiteStorage("widget_state.db"))

Options

• path (str) - The SQLite file, created if missing. Required.

Source code in dachshund/storage/sqlite.py

def __init__(self, path: str) -> None:
    self._connection, self._executor = db.setup(path, MIGRATIONS)
    self._closed = False

close

close()

Close the file. Calling it again does nothing.

The dashboard never closes a storage, close it when you're done with it.

Source code in dachshund/storage/sqlite.py

def close(self) -> None:
    """Close the file. Calling it again does nothing.

    The dashboard never closes a storage, close it when you're done with it.
    """
    if self._closed:
        return

    self._closed = True
    db.close(self._connection, self._executor)

save async

save(key, value)

Replace whatever was kept under a key.

Source code in dachshund/storage/sqlite.py

async def save(self, key: str, value: dict[str, Any]) -> None:
    """Replace whatever was kept under a key."""
    loop = asyncio.get_running_loop()
    await loop.run_in_executor(self._executor, self._save, key, value)

load async

load(key)

What was last saved under a key, or None if nothing was.

Source code in dachshund/storage/sqlite.py

async def load(self, key: str) -> dict[str, Any] | None:
    """What was last saved under a key, or `None` if nothing was."""
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(self._executor, self._load, key)

dachshund.storage.protocol.SyncStorage

A Storage for a backend with synchronous calls, such as an ORM.

Subclass and write _save(key, value) and _load(key). Each call runs in a thread, off the event loop, so the backend must be safe to call from any thread.

Example

from dachshund.storage import SyncStorage


class FileStorage(SyncStorage):
    def __init__(self, path: str) -> None:
        self.path = path

    def _save(self, key, value): ...

    def _load(self, key): ...

save async

save(key, value)

Replace whatever was kept under a key.

Source code in dachshund/storage/protocol.py

async def save(self, key: str, value: dict[str, Any]) -> None:
    """Replace whatever was kept under a key."""
    await asyncio.to_thread(self._save, key, value)

load async

load(key)

What was last saved under a key, or None if nothing was.

Source code in dachshund/storage/protocol.py

async def load(self, key: str) -> dict[str, Any] | None:
    """What was last saved under a key, or `None` if nothing was."""
    return await asyncio.to_thread(self._load, key)