1 """ sqlite event store """
3 from sqlite3 import connect, OperationalError
4 from typing import Any, List, Tuple
6 __all__ = "fetch", "initdb", "stow"
10 SCHEMA = """create table if not exists events (
13 peeraddr text not null,
14 is_incoming int not null default TRUE,
20 def initdb(dbname: str) -> None:
25 """alter table events add column
26 is_incoming int not null default TRUE"""
28 except OperationalError:
32 def stow(**kwargs: Any) -> None:
35 k: kwargs[k] if k in kwargs else v
37 ("is_incoming", True),
45 assert len(kwargs) <= len(parms)
47 """insert or ignore into events
48 (tstamp, imei, peeraddr, proto, packet, is_incoming)
50 (:when, :imei, :peeraddr, :proto, :packet, :is_incoming)
58 imei: str, matchlist: List[Tuple[bool, str]], backlog: int
59 ) -> List[Tuple[bool, float, bytes]]:
60 # matchlist is a list of tuples (is_incoming, proto)
61 # returns a list of tuples (is_incoming, timestamp, packet)
63 selector = " or ".join(
64 (f"(is_incoming = ? and proto = ?)" for _ in range(len(matchlist)))
68 f"""select is_incoming, tstamp, packet from events
69 where ({selector}) and imei = ?
70 order by tstamp desc limit ?""",
71 tuple(item for sublist in matchlist for item in sublist)
76 return list(reversed(result))