gps303/evstore.py

   1 """ sqlite event store """
   2
   3 from sqlite3 import connect, OperationalError
   4 from typing import Any, List, Tuple
   5
   6 __all__ = "fetch", "initdb", "stow"
   7
   8 DB = None
   9
  10 SCHEMA = """create table if not exists events (
  11     tstamp real not null,
  12     imei text,
  13     peeraddr text not null,
  14     is_incoming int not null default TRUE,
  15     proto int not null,
  16     packet blob
  17 )"""
  18
  19
  20 def initdb(dbname: str) -> None:
  21     global DB
  22     DB = connect(dbname)
  23     try:
  24         DB.execute(
  25             """alter table events add column
  26                 is_incoming int not null default TRUE"""
  27         )
  28     except OperationalError:
  29         DB.execute(SCHEMA)
  30
  31
  32 def stow(**kwargs: Any) -> None:
  33     assert DB is not None
  34     parms = {
  35         k: kwargs[k] if k in kwargs else v
  36         for k, v in (
  37             ("is_incoming", True),
  38             ("peeraddr", None),
  39             ("when", 0.0),
  40             ("imei", None),
  41             ("proto", -1),
  42             ("packet", b""),
  43         )
  44     }
  45     assert len(kwargs) <= len(parms)
  46     DB.execute(
  47         """insert or ignore into events
  48                 (tstamp, imei, peeraddr, proto, packet, is_incoming)
  49                 values
  50                 (:when, :imei, :peeraddr, :proto, :packet, :is_incoming)
  51         """,
  52         parms,
  53     )
  54     DB.commit()
  55
  56
  57 def fetch(
  58     imei: str, matchlist: List[Tuple[bool, int]], 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)
  62     assert DB is not None
  63     selector = " or ".join(
  64         (f"(is_incoming = ? and proto = ?)" for _ in range(len(matchlist)))
  65     )
  66     cur = DB.cursor()
  67     cur.execute(
  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)
  72         + (imei, backlog),
  73     )
  74     result = list(cur)
  75     cur.close()
  76     return list(reversed(result))