1 """ sqlite event store """
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
8 __all__ = "fetch", "initdb", "stow", "stowloc"
13 """create table if not exists events (
16 peeraddr text not null,
17 is_incoming int not null default TRUE,
21 """create table if not exists reports (
23 devtime text not null,
32 def initdb(dbname: str) -> None:
40 def stow(**kwargs: Any) -> None:
43 k: kwargs[k] if k in kwargs else v
45 ("is_incoming", True),
53 assert len(kwargs) <= len(parms)
55 """insert or ignore into events
56 (tstamp, imei, peeraddr, proto, packet, is_incoming)
58 (:when, :imei, :peeraddr, :proto, :packet, :is_incoming)
65 def stowloc(**kwargs: Dict[str, Any]) -> None:
68 k: kwargs.pop(k) if k in kwargs else v
71 ("devtime", str(datetime.now())),
77 parms["remainder"] = dumps(kwargs)
79 """insert or ignore into reports
80 (imei, devtime, accuracy, latitude, longitude, remainder)
82 (:imei, :devtime, :accuracy, :latitude, :longitude, :remainder)
89 def fetch(imei: str, backlog: int) -> List[Dict[str, Any]]:
93 """select imei, devtime, accuracy, latitude, longitude, remainder
94 from reports where imei = ?
95 order by devtime desc limit ?""",
101 remainder = loads(dic.pop("remainder"))
102 dic.update(remainder)
105 return list(reversed(result))