X-Git-Url: http://average.org/gitweb/?a=blobdiff_plain;f=gps303%2Fevstore.py;h=2b23e8e92ef98e4c5ff21a584be786df179859ad;hb=f32d7fdd5312dce5b8df09f9a3360a4f9856ad8f;hp=a52a64c1e5ad468676349ad299e6fb14b174eceb;hpb=311d3cc7b0692e66edb9b9bb9285b2bfc094d571;p=loctrkd.git diff --git a/gps303/evstore.py b/gps303/evstore.py index a52a64c..2b23e8e 100644 --- a/gps303/evstore.py +++ b/gps303/evstore.py @@ -1,6 +1,7 @@ """ sqlite event store """ from sqlite3 import connect, OperationalError +from typing import Any, List, Tuple __all__ = "fetch", "initdb", "stow" @@ -16,7 +17,7 @@ SCHEMA = """create table if not exists events ( )""" -def initdb(dbname): +def initdb(dbname: str) -> None: global DB DB = connect(dbname) try: @@ -28,7 +29,7 @@ def initdb(dbname): 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 @@ -53,7 +54,9 @@ def stow(**kwargs): DB.commit() -def fetch(imei, matchlist, backlog): +def fetch( + imei: str, matchlist: List[Tuple[bool, int]], 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