Skip to content
hikari-risa hikari-risa

Writing your own store

A store is six methods. Write them against what you already run, and verify_store tells you whether risa can rely on them.

The six methods

risa.Store is a protocol, and the reference has each method in full.

Method What it has to do
get_versioned(key) Return the bytes last written, unchanged, with the entry's version, as a risa.VersionedEntry, or None if the entry is missing or expired
put(key, value, *, ttl) Write whether or not an entry exists, and change the version
put_if_version(key, value, *, expected, ttl) Write only if the entry is at version expected, in one atomic step, and return the new version. Otherwise write nothing and return None
delete(key) Remove the entry, if there is one
touch(key, *, ttl) Restart the entry's lifetime, leave its data and version alone, and leave a missing key missing
lock(key, *, timeout=...) Return an async context manager that holds the key exclusively and leaves other keys free. It raises risa.LockTimeoutError after waiting timeout seconds

ttl is in seconds, fractions included. None means no expiry and clears any earlier one.

A missing entry is never an error. Let every other failure raise: risa reports it as StoreUnavailableError, so an outage cannot pass for an expired view.

A minimal store

DictStore keeps its entries in a dict. put_if_version decides whose write lands when two clicks race:

async def put_if_version(
    self,
    key: str,
    value: bytes,
    *,
    expected: int,
    ttl: float | None,
) -> int | None:
    entry = self._live(key)
    if entry is None or entry.version != expected:
        return None
    written = self._new_entry(value, ttl)
    self._entries[key] = written
    return written.version

No await sits between the compare and the write here, so nothing can run between them. A real backend needs one step of its own for this, as RedisStore has a Lua script and PostgresStore one UPDATE.

The lock is one asyncio.Lock per key:

@contextlib.asynccontextmanager
async def lock(
    self,
    key: str,
    *,
    timeout: float = risa.DEFAULT_LOCK_TIMEOUT,
) -> collections.abc.AsyncGenerator[None]:
    lock = self._locks.setdefault(key, asyncio.Lock())
    try:
        async with asyncio.timeout(timeout):
            await lock.acquire()
    except TimeoutError as exc:
        raise risa.LockTimeoutError(key, timeout) from exc
    try:
        yield
    finally:
        lock.release()

risa calls lock without await, as asynccontextmanager allows. A client takes the store like any other:

client = risa.client_from_app(bot, store=DictStore())
client.add_view(Tally)
Full file 
"""A store of your own, and the test that holds it to what risa relies on."""

from __future__ import annotations

import asyncio
import contextlib
import itertools
import os
import time
import typing

import hikari
import msgspec

import risa
from risa import ui
from risa.store.testing import verify_store

if typing.TYPE_CHECKING:
    import collections.abc


class _Entry(msgspec.Struct):
    data: bytes
    version: int
    expires_at: float | None


class DictStore(risa.Store):
    def __init__(self) -> None:
        self._entries: dict[str, _Entry] = {}
        self._locks: dict[str, asyncio.Lock] = {}
        self._versions = itertools.count(1)

    async def get_versioned(self, key: str) -> risa.VersionedEntry | None:
        entry = self._live(key)
        if entry is None:
            return None
        return risa.VersionedEntry(data=entry.data, version=entry.version)

    async def put(self, key: str, value: bytes, *, ttl: float | None) -> None:
        self._entries[key] = self._new_entry(value, ttl)

    async def put_if_version(
        self,
        key: str,
        value: bytes,
        *,
        expected: int,
        ttl: float | None,
    ) -> int | None:
        entry = self._live(key)
        if entry is None or entry.version != expected:
            return None
        written = self._new_entry(value, ttl)
        self._entries[key] = written
        return written.version

    async def delete(self, key: str) -> None:
        self._entries.pop(key, None)

    async def touch(self, key: str, *, ttl: float) -> None:
        entry = self._live(key)
        if entry is not None:
            entry.expires_at = time.monotonic() + ttl

    @contextlib.asynccontextmanager
    async def lock(
        self,
        key: str,
        *,
        timeout: float = risa.DEFAULT_LOCK_TIMEOUT,
    ) -> collections.abc.AsyncGenerator[None]:
        lock = self._locks.setdefault(key, asyncio.Lock())
        try:
            async with asyncio.timeout(timeout):
                await lock.acquire()
        except TimeoutError as exc:
            raise risa.LockTimeoutError(key, timeout) from exc
        try:
            yield
        finally:
            lock.release()

    def _live(self, key: str) -> _Entry | None:
        entry = self._entries.get(key)
        if entry is None or entry.expires_at is None:
            return entry
        if entry.expires_at > time.monotonic():
            return entry
        del self._entries[key]
        return None

    def _new_entry(self, value: bytes, ttl: float | None) -> _Entry:
        expires_at = None if ttl is None else time.monotonic() + ttl
        return _Entry(value, next(self._versions), expires_at)


async def test_dict_store() -> None:
    await verify_store(DictStore())


@risa.register(name="tally", state=risa.InStore(ttl=3600))
class Tally(risa.View):
    count: int = 0

    def render(self) -> ui.Layout:
        return ui.Row(ui.Button(self.add, label=f"Count: {self.count}"))

    @risa.handler
    async def add(self, ctx: risa.ComponentContext) -> None:
        self.count += 1
        await ctx.rerender()


def main() -> None:
    bot = hikari.GatewayBot(
        os.environ["DISCORD_TOKEN"],
        intents=hikari.Intents.ALL_UNPRIVILEGED | hikari.Intents.MESSAGE_CONTENT,
    )
    client = risa.client_from_app(bot, store=DictStore())
    client.add_view(Tally)

    @bot.listen()
    async def on_message(event: hikari.GuildMessageCreateEvent) -> None:
        if event.content == "!tally":
            rendered = await client.build(Tally())
            await rendered.send_to(bot.rest, event.channel_id)

    bot.run()


if __name__ == "__main__":
    main()

Test it

Add verify_store to your tests, with a store connected to a real instance of your backend.

async def test_dict_store() -> None:
    await verify_store(DictStore())

This one needs pytest-asyncio with asyncio_mode = "auto". In a script, asyncio.run(verify_store(DictStore())) does the same. A failed check makes it raise StoreConformanceError, which lists every failure.

Try it.

Take or entry.version != expected out of put_if_version. verify_store now names a_stale_cas_is_refused and racing_cas_writes_have_one_winner.

Stores that processes share

verify_store runs in one process, so a lock that excludes only its own process passes. For several processes, lock has to exclude all of them. Without that, two clicks run at once, and one of them gets StateConflictError.

DictStore is a teaching example, for one process only. It gets no startup warning, and it keeps every lock and every expired entry nobody reads, so a bot that runs on it grows until it restarts.