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(
  24             """alter table events add column
  25                 is_incoming int not null default TRUE"""
  26         )
  27     except OperationalError:
  28         DB.execute(SCHEMA)
  29
  30
  31 def stow(**kwargs):
  32     assert DB is not None
  33     parms = {
  34         k: kwargs[k] if k in kwargs else v
  35         for k, v in (
  36             ("is_incoming", True),
  37             ("peeraddr", None),
  38             ("when", 0.0),
  39             ("imei", None),
  40             ("proto", -1),
  41             ("packet", b""),
  42         )
  43     }
  44     assert len(kwargs) <= len(parms)
  45     DB.execute(
  46         """insert or ignore into events
  47                 (tstamp, imei, peeraddr, proto, packet, is_incoming)
  48                 values
  49                 (:when, :imei, :peeraddr, :proto, :packet, :is_incoming)
  50         """,
  51         parms,
  52     )
  53     DB.commit()
  54
  55
  56 def fetch(imei, protos, backlog):
  57     assert DB is not None
  58     protosel = ", ".join(["?" for _ in range(len(protos))])
  59     cur = DB.cursor()
  60     cur.execute(
  61         f"""select packet from events
  62                     where proto in ({protosel}) and imei = ?
  63                     order by tstamp desc limit ?""",
  64         protos + (imei, backlog),
  65     )
  66     result = [row[0] for row in cur]
  67     cur.close()
  68     return result