loctrkd/evstore.py

   1 """ sqlite event store """
   2
   3 from datetime import datetime
   4 from json import dumps, loads
   5 from sqlite3 import connect, OperationalError, Row
   6 from typing import Any, Dict, List, Tuple
   7
   8 __all__ = "fetch", "initdb", "stow", "stowloc"
   9
  10 DB = None
  11
  12 SCHEMA = (
  13     """create table if not exists events (
  14     tstamp real not null,
  15     imei text,
  16     peeraddr text not null,
  17     is_incoming int not null default TRUE,
  18     proto text not null,
  19     packet blob
  20 )""",
  21     """create table if not exists reports (
  22     imei text,
  23     devtime text not null,
  24     accuracy real,
  25     latitude real,
  26     longitude real,
  27     remainder text
  28 )""",
  29 )
  30
  31
  32 def initdb(dbname: str) -> None:
  33     global DB
  34     DB = connect(dbname)
  35     DB.row_factory = Row
  36     for stmt in SCHEMA:
  37         DB.execute(stmt)
  38
  39
  40 def stow(**kwargs: Any) -> None:
  41     assert DB is not None
  42     parms = {
  43         k: kwargs[k] if k in kwargs else v
  44         for k, v in (
  45             ("is_incoming", True),
  46             ("peeraddr", None),
  47             ("when", 0.0),
  48             ("imei", None),
  49             ("proto", "UNKNOWN"),
  50             ("packet", b""),
  51         )
  52     }
  53     assert len(kwargs) <= len(parms)
  54     DB.execute(
  55         """insert or ignore into events
  56                 (tstamp, imei, peeraddr, proto, packet, is_incoming)
  57                 values
  58                 (:when, :imei, :peeraddr, :proto, :packet, :is_incoming)
  59         """,
  60         parms,
  61     )
  62     DB.commit()
  63
  64
  65 def stowloc(**kwargs: Dict[str, Any]) -> None:
  66     assert DB is not None
  67     parms = {
  68         k: kwargs.pop(k) if k in kwargs else v
  69         for k, v in (
  70             ("imei", None),
  71             ("devtime", str(datetime.now())),
  72             ("accuracy", None),
  73             ("latitude", None),
  74             ("longitude", None),
  75         )
  76     }
  77     parms["remainder"] = dumps(kwargs)
  78     DB.execute(
  79         """insert or ignore into reports
  80                 (imei, devtime, accuracy, latitude, longitude, remainder)
  81                 values
  82                 (:imei, :devtime, :accuracy, :latitude, :longitude, :remainder)
  83         """,
  84         parms,
  85     )
  86     DB.commit()
  87
  88
  89 def fetch(imei: str, backlog: int) -> List[Dict[str, Any]]:
  90     assert DB is not None
  91     cur = DB.cursor()
  92     cur.execute(
  93         """select imei, devtime, accuracy, latitude, longitude, remainder
  94                     from reports where imei = ?
  95                     order by devtime desc limit ?""",
  96         (imei, backlog),
  97     )
  98     result = []
  99     for row in cur:
 100         dic = dict(row)
 101         remainder = loads(dic.pop("remainder"))
 102         dic.update(remainder)
 103         result.append(dic)
 104     cur.close()
 105     return list(reversed(result))