yera.ui.sessions_store.sqlite_store

Module for the SQLite-based session store.

Symbols

class SQLiteSessionStore — SQLite-based implementation of BaseSessionStore.

SQLiteSessionStore

Inherits: BaseSessionStore

SQLite-based implementation of BaseSessionStore.

The store creates a tiny SQLite database (WAL mode, foreign keys enabled) in the user's home directory unless another path is supplied. It provides CRUD operations for session metadata and an event log stored as JSON strings.

Methods

create — Create a new session record for *app_id*.
has — Return ``True`` if a session with *session_id* exists in the store.
has_events — Check whether any events are stored for *session_id*.
get — Retrieve the :class:`SessionRecord` for *session_id*.
list — Return all stored sessions ordered by most recent update time.
update — Update metadata for *session_id*.
append_event — Append an event to the log of *session_id*.
read_events — Yield all events for *session_id* in chronological order.
__enter__ — Enter a runtime context and initialise the SQLite store.
__exit__ — Exit the runtime context, resetting the opened flag.

SQLiteSessionStore.create

create(
    app_id: str,
) → SessionRecord

Create a new session record for app_id.

Parameters

app_id
type: str

Identifier of the application that owns the session.

Returns

type: SessionRecord

The newly created :class:SessionRecord.

SQLiteSessionStore.has

has(
    session_id: str,
) → bool

Return True if a session with session_id exists in the store.

Parameters

session_id
type: str

The UUID of the session to look up.

Returns

type: bool

True if the session is present, otherwise False.

SQLiteSessionStore.has_events

has_events(
    session_id: str,
) → bool

Check whether any events are stored for session_id.

Parameters

session_id
type: str

The UUID of the session whose event log is queried.

Returns

type: bool

True if at least one event exists, otherwise False.

SQLiteSessionStore.get

get(
    session_id: str,
) → SessionRecord

Retrieve the :class:SessionRecord for session_id.

Parameters

session_id
type: str

The UUID of the desired session.

Returns

type: SessionRecord

The corresponding :class:SessionRecord.

Raises

KeyError

If no such session exists.

SQLiteSessionStore.list

list() → list[SessionRecord]

Return all stored sessions ordered by most recent update time.

Returns

type: list[SessionRecord]

A list of :class:SessionRecord objects sorted descending by updated_at.

SQLiteSessionStore.update

update(
    session_id: str,
    title: str | None = None,
    status: SessionStatus | None = None,
) → None

Update metadata for session_id.

Parameters

session_id
type: str

The UUID of the session to modify.

title
type: str | None = None

Optional new title. If omitted, only the updated_at timestamp is refreshed.

status
type: SessionStatus | None = None

An optional new lifecycle status for the session.

Raises

KeyError

If the specified session does not exist.

SQLiteSessionStore.append_event

append_event(
    session_id: str,
    event: OutputEvent | InputEvent,
    update: SessionRecordUpdate | None = None,
) → None

Append an event to the log of session_id.

Parameters

session_id
type: str

The UUID of the target session.

event
type: OutputEvent | InputEvent

An event to store.

update
type: SessionRecordUpdate | None = None

Optional session metadata changes to apply atomically.

Raises

KeyError

If the session does not exist.

SQLiteSessionStore.read_events

read_events(
    session_id: str,
) → Iterator[OutputEvent | InputEvent]

Yield all events for session_id in chronological order.

Parameters

session_id
type: str

The UUID of the session whose events should be streamed.

Raises

KeyError

If the session does not exist. Raised eagerly when read_events is called.

ValueError

If an unknown event type is encountered. Raised lazily during iteration, not at call time.

SQLiteSessionStore.__enter__

__enter__() → Self

Enter a runtime context and initialise the SQLite store.

Returns

type: Self

self the open SQLiteSessionStore instance.

SQLiteSessionStore.__exit__

__exit__(
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) → None

Exit the runtime context, resetting the opened flag.