ms
22 lines
1.3 KiB
Python
22 lines
1.3 KiB
Python
import json,sqlite3,threading
|
|
from pathlib import Path
|
|
class Outbox:
|
|
def __init__(self,path,limit=1000000):
|
|
Path(path).parent.mkdir(parents=True,exist_ok=True)
|
|
self.db=sqlite3.connect(path,check_same_thread=False)
|
|
self.db.execute('PRAGMA journal_mode=WAL');self.db.execute('PRAGMA synchronous=FULL')
|
|
self.db.execute('CREATE TABLE IF NOT EXISTS outbox (seq INTEGER PRIMARY KEY AUTOINCREMENT, id TEXT UNIQUE, body TEXT NOT NULL)')
|
|
self.db.commit();self.lock=threading.Lock();self.limit=limit
|
|
def add(self,event):
|
|
with self.lock:
|
|
if self.db.execute('SELECT COUNT(*) FROM outbox').fetchone()[0]>=self.limit:
|
|
raise RuntimeError('Outbox full: collector stopped to avoid silent data loss')
|
|
with self.db:self.db.execute('INSERT OR IGNORE INTO outbox(id,body) VALUES (?,?)',(event['id'],json.dumps(event,allow_nan=False)))
|
|
def batch(self,limit=200):
|
|
with self.lock:return [(r[0],json.loads(r[1])) for r in self.db.execute('SELECT seq,body FROM outbox ORDER BY seq LIMIT ?',(limit,))]
|
|
def ack(self,ids):
|
|
with self.lock,self.db:self.db.executemany('DELETE FROM outbox WHERE seq=?',[(i,) for i in ids])
|
|
def count(self):
|
|
with self.lock:return self.db.execute('SELECT COUNT(*) FROM outbox').fetchone()[0]
|
|
def close(self): self.db.close()
|