this repo has no description
edavis.dev
1import apsw
2
3class FirehoseManager:
4 def __init__(self, fname='firehose.db'):
5 self.db_cnx = apsw.Connection(fname)
6 self.db_cnx.pragma('journal_mode', 'WAL')
7 with self.db_cnx:
8 self.db_cnx.execute("create table if not exists firehose(key text unique, value text)")
9
10 def get_sequence_number(self):
11 cur = self.db_cnx.execute("select * from firehose where key = 'seq'")
12 row = cur.fetchone()
13 if row is None:
14 return None
15 (key, value) = row
16 return int(value)
17
18 def set_sequence_number(self, value):
19 with self.db_cnx:
20 self.db_cnx.execute(
21 "insert into firehose (key, value) values ('seq', :value) on conflict(key) do update set value = :value",
22 dict(value=value)
23 )