social media crossposting tool. 3rd time's the charm
zenfyr.dev
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
17
18 def close(self):
19 for c in self._conns:
20 c.close()
21
22def get_conn(db: Path) -> sqlite3.Connection:
23 conn = sqlite3.connect(db, autocommit=True, check_same_thread=False)
24 conn.row_factory = sqlite3.Row
25 _ = conn.executescript("""
26 PRAGMA journal_mode = WAL;
27 PRAGMA mmap_size = 134217728;
28 PRAGMA cache_size = 4000;
29 PRAGMA synchronous = NORMAL;
30 PRAGMA foreign_keys = ON;
31 """)
32 return conn