37 lines
1.2 KiB
Python
37 lines
1.2 KiB
Python
"""SQLite connection plumbing shared by the database layer."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
from datetime import datetime
|
|
|
|
# Per-connection busy timeout (ms) so concurrent writers wait instead of
|
|
# failing with "database is locked".
|
|
SQLITE_BUSY_TIMEOUT_MS = 5000
|
|
|
|
|
|
def _parse_timestamp(value: str | bytes) -> datetime:
|
|
"""Decode a stored TIMESTAMP value into a naive datetime."""
|
|
if isinstance(value, bytes):
|
|
value = value.decode("utf-8")
|
|
return datetime.fromisoformat(value)
|
|
|
|
|
|
sqlite3.register_converter("TIMESTAMP", _parse_timestamp)
|
|
|
|
|
|
def connect(db_path: str) -> sqlite3.Connection:
|
|
"""Open a SQLite connection configured for concurrent access.
|
|
|
|
WAL journaling is persistent (set once per database file); the busy
|
|
timeout is per-connection, so it is applied on every connection here.
|
|
``PARSE_DECLTYPES`` plus the registered ``TIMESTAMP`` converter decode
|
|
declared ``TIMESTAMP`` columns into ``datetime`` objects instead of
|
|
raw strings.
|
|
|
|
"""
|
|
conn = sqlite3.connect(db_path, detect_types=sqlite3.PARSE_DECLTYPES)
|
|
conn.execute("PRAGMA journal_mode=WAL")
|
|
conn.execute(f"PRAGMA busy_timeout={SQLITE_BUSY_TIMEOUT_MS}")
|
|
return conn
|