gps303/evstore.py

   1 """ sqlite event store """
   2
   3 from sqlite3 import connect, OperationalError
   4
   5 __all__ = "fetch", "initdb", "stow"
   6
   7 DB = None
   8
   9 SCHEMA = """create table if not exists events (
  10     tstamp real not null,
  11     imei text,
  12     peeraddr text not null,
  13     is_incoming int not null default TRUE,
  14     proto int not null,
  15     packet blob
  16 )"""
  17
  18
  19 def initdb(dbname):
  20     global DB
  21     DB = connect(dbname)
  22     try:
  23         DB.execute("""alter table events add column
  24                 is_incoming int not null default TRUE""")
  25     except OperationalError:
  26         DB.execute(SCHEMA)
  27
  28
  29 def stow(**kwargs):
  30     assert DB is not None
  31     parms = {
  32         k: kwargs[k] if k in kwargs else v
  33         for k, v in (
  34             ("peeraddr", None),
  35             ("when", 0.0),
  36             ("imei", None),
  37             ("proto", -1),
  38             ("packet", b""),
  39         )
  40     }
  41     assert len(kwargs) <= len(parms)
  42     DB.execute(
  43         """insert or ignore into events
  44                 (tstamp, imei, peeraddr, proto, packet)
  45                 values
  46                 (:when, :imei, :peeraddr, :proto, :packet)
  47         """,
  48         parms,
  49     )
  50     DB.commit()
  51
  52 def fetch(imei, protos, backlog):
  53     assert DB is not None
  54     protosel = ", ".join(["?" for _ in range(len(protos))])
  55     cur = DB.cursor()
  56     cur.execute(f"""select packet from events
  57                     where proto in ({protosel}) and imei = ?
  58                     order by tstamp desc limit ?""",
  59                 protos + (imei, backlog))
  60     result = [row[0] for row in cur]
  61     cur.close()
  62     return result