Something went wrong. Try again.
social media crossposting tool. 3rd time's the charm
mastodon misskey crossposting bluesky
Something went wrong. Try again.
12345678910111213141516171819202122232425262728293031323334import sqlite3import threadingfrom pathlib import Path
class DatabasePool: def __init__(self, db: Path) -> None: self.db: Path = db self._local: threading.local = threading.local() self._conns: list[sqlite3.Connection] = []
def get_conn(self) -> sqlite3.Connection: if getattr(self._local, "conn", None) is None: self._local.conn = get_conn(self.db) self._conns.append(self._local.conn) return self._local.conn # type: ignore[no-any-return]
def close(self) -> None: for c in self._conns: c.close()
def get_conn(db: Path) -> sqlite3.Connection: conn = sqlite3.connect(db, autocommit=True, check_same_thread=False) conn.row_factory = sqlite3.Row _ = conn.executescript(""" PRAGMA journal_mode = WAL; PRAGMA mmap_size = 134217728; PRAGMA cache_size = 4000; PRAGMA synchronous = NORMAL; PRAGMA foreign_keys = ON; """) return conn