Keyed Widget

KeyedWidget is the base class of a widget whose state is one entry per key, like Table and BarChart.

from dachshund.widgets import KeyedWidget


class Leaderboard(KeyedWidget):
    type = "leaderboard"

    def accumulate(self, event):
        self.state.upsert(event.payload["user"], event.payload["score"])

dachshund.widgets.state.KeyedWidget dataclass

KeyedWidget(name, *, source='', title=None, actions=list(), action_args=dict(), persist=False)

Bases: PersistableWidget

A PersistableWidget whose state is one entry per key.

snapshot(), restore() and sweep() are provided. A subclass writes accumulate() and reads self.state.entries in render(). Table and BarChart are built this way.

Example

class Leaderboard(KeyedWidget):
    type = "leaderboard"

    def accumulate(self, event):
        self.state.upsert(event.payload["user"], event.payload["score"])

state class-attribute instance-attribute

state = field(default_factory=KeyedState, init=False, repr=False, compare=False)

snapshot

snapshot()

The kept entries.

Source code in dachshund/widgets/state.py

def snapshot(self) -> dict[str, Any]:
    """The kept entries."""
    return self.state.snapshot()

restore

restore(data)

Pick up where the last persisted snapshot left off.

Source code in dachshund/widgets/state.py

def restore(self, data: dict[str, Any]) -> None:
    """Pick up where the last persisted snapshot left off."""
    self.state.restore(data)

sweep

sweep(before)

Remove every entry not seen since before, a time.time() value.

To resync after downtime, emit every current entry again, then sweep.

Example

started = time.time()

for user in current_users():
    dachshund.emit("sessions", user=user.name, ip=user.ip)

sessions.sweep(before=started)

Source code in dachshund/widgets/state.py

def sweep(self, before: float) -> None:
    """Remove every entry not seen since `before`, a `time.time()` value.

    To resync after downtime, emit every current entry again, then sweep.

    ## Example

    ```python
    started = time.time()

    for user in current_users():
        dachshund.emit("sessions", user=user.name, ip=user.ip)

    sessions.sweep(before=started)
    ```
    """
    self.state.sweep(before)