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             ("is_incoming", True),
  35             ("peeraddr", None),
  36             ("when", 0.0),
  37             ("imei", None),
  38             ("proto", -1),
  39             ("packet", b""),
  40         )
  41     }
  42     assert len(kwargs) <= len(parms)
  43     DB.execute(
  44         """insert or ignore into events
  45                 (tstamp, imei, peeraddr, proto, packet, is_incoming)
  46                 values
  47                 (:when, :imei, :peeraddr, :proto, :packet, :is_incoming)
  48         """,
  49         parms,
  50     )
  51     DB.commit()
  52
  53 def fetch(imei, protos, backlog):
  54     assert DB is not None
  55     protosel = ", ".join(["?" for _ in range(len(protos))])
  56     cur = DB.cursor()
  57     cur.execute(f"""select packet from events
  58                     where proto in ({protosel}) and imei = ?
  59                     order by tstamp desc limit ?""",
  60                 protos + (imei, backlog))
  61     result = [row[0] for row in cur]
  62     cur.close()
  63     return result