zenfyr.dev
social media crossposting tool. 3rd time's the charm
mastodon
misskey
crossposting
bluesky
1import sqlite3
2import threading
3from pathlib import Path
4
5
6class DatabasePool:
7 def __init__(self, db: Path) -> None:
8 self.db: Path = db
9 self._local: threading.local = threading.local()
10 self._conns: list[sqlite3.Connection] = []
11
12 def get_conn(self) -> sqlite3.Connection:
13 if getattr(self._local, "conn", None) is None:
14 self._local.conn = get_conn(self.db)
15 self._conns.append(self._local.conn)
16 return self._local.conn # type: ignore[no-any-return]
17
18 def close(self) -> None:
19 for c in self._conns:
20 c.close()
21
22
23def get_conn(db: Path) -> sqlite3.Connection:
24 conn = sqlite3.connect(db, autocommit=True, check_same_thread=False)
25 conn.row_factory = sqlite3.Row
26 _ = conn.executescript("""
27 PRAGMA journal_mode = WAL;
28 PRAGMA mmap_size = 134217728;
29 PRAGMA cache_size = 4000;
30 PRAGMA synchronous = NORMAL;
31 PRAGMA foreign_keys = ON;
32 """)
33 return conn