Joseph Doherty
28 lines
666 B
Python
28 lines
666 B
Python
from __future__ import annotations
|
|
from collections.abc import Callable
|
|
from sqlite3 import Connection
|
|
from .log import Event, read_events
|
|
|
|
Handler = Callable[[Connection, Event], None]
|
|
_REGISTRY: dict[str, Handler] = {}
|
|
|
|
|
|
def on(kind: str):
|
|
def deco(fn: Handler) -> Handler:
|
|
_REGISTRY[kind] = fn
|
|
return fn
|
|
return deco
|
|
|
|
|
|
def project(conn: Connection, branch_id: int = 1) -> None:
|
|
for event in read_events(conn, branch_id=branch_id):
|
|
h = _REGISTRY.get(event.kind)
|
|
if h:
|
|
h(conn, event)
|
|
|
|
|
|
def apply_event(conn: Connection, event: Event) -> None:
|
|
h = _REGISTRY.get(event.kind)
|
|
if h:
|
|
h(conn, event)
|