""" sqlite event store """
from sqlite3 import connect, OperationalError
+from typing import Any, List, Tuple
__all__ = "fetch", "initdb", "stow"
imei text,
peeraddr text not null,
is_incoming int not null default TRUE,
- proto int not null,
+ proto text not null,
packet blob
)"""
-def initdb(dbname):
+def initdb(dbname: str) -> None:
global DB
DB = connect(dbname)
try:
DB.execute(SCHEMA)
-def stow(**kwargs):
+def stow(**kwargs: Any) -> None:
assert DB is not None
parms = {
k: kwargs[k] if k in kwargs else v
("peeraddr", None),
("when", 0.0),
("imei", None),
- ("proto", -1),
+ ("proto", "UNKNOWN"),
("packet", b""),
)
}
DB.commit()
-def fetch(imei, matchlist, backlog):
+def fetch(
+ imei: str, matchlist: List[Tuple[bool, str]], backlog: int
+) -> List[Tuple[bool, float, bytes]]:
# matchlist is a list of tuples (is_incoming, proto)
# returns a list of tuples (is_incoming, timestamp, packet)
assert DB is not None