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, Optional, 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,
29 """create table if not exists pmodmap (
30 imei text not null unique,
36 def initdb(dbname: str) -> None:
44 def stow(**kwargs: Any) -> None:
47 k: kwargs[k] if k in kwargs else v
49 ("is_incoming", True),
57 assert len(kwargs) <= len(parms)
59 """insert or ignore into events
60 (tstamp, imei, peeraddr, proto, packet, is_incoming)
62 (:when, :imei, :peeraddr, :proto, :packet, :is_incoming)
69 def stowloc(**kwargs: Dict[str, Any]) -> None:
72 k: kwargs.pop(k) if k in kwargs else v
75 ("devtime", str(datetime.now())),
81 parms["remainder"] = dumps(kwargs)
83 """insert or ignore into reports
84 (imei, devtime, accuracy, latitude, longitude, remainder)
86 (:imei, :devtime, :accuracy, :latitude, :longitude, :remainder)
93 def stowpmod(imei: str, pmod: str) -> None:
96 """insert or replace into pmodmap
97 (imei, pmod) values (:imei, :pmod)
99 {"imei": imei, "pmod": pmod},
104 def fetch(imei: str, backlog: int) -> List[Dict[str, Any]]:
105 assert DB is not None
108 """select imei, devtime, accuracy, latitude, longitude, remainder
109 from reports where imei = ?
110 order by devtime desc limit ?""",
116 remainder = loads(dic.pop("remainder"))
117 dic.update(remainder)
120 return list(reversed(result))
123 def fetchpmod(imei: str) -> Optional[Any]:
124 assert DB is not None
127 cur.execute("select pmod from pmodmap where imei = ?", (imei,))
128 result = cur.fetchone()