feat: add database module with connection factory and transactions

Implements Step 1.2:
- constants.py: END_OF_TIME_UTC = 32503680000000 (year 3000 sentinel)
- db.py: get_db() with pragmas (WAL, synchronous=FULL, foreign_keys, busy_timeout)
- db.py: transaction() context manager with BEGIN IMMEDIATE

Includes 12 TDD tests for pragmas, commit/rollback, and concurrent writes.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
2025-12-27 17:58:29 +00:00
parent 35e832698e
commit d8910d6571
5 changed files with 258 additions and 5 deletions

View File

@@ -0,0 +1,6 @@
# ABOUTME: Application-wide constants and sentinel values.
# ABOUTME: Defines END_OF_TIME_UTC for interval indexes.
# Sentinel constant for unbounded end_utc in interval indexes.
# Represents January 1, 3000 00:00:00 UTC as milliseconds since Unix epoch.
END_OF_TIME_UTC: int = 32503680000000

76
src/animaltrack/db.py Normal file
View File

@@ -0,0 +1,76 @@
# ABOUTME: Database connection factory and transaction management.
# ABOUTME: Configures SQLite pragmas and provides BEGIN IMMEDIATE transactions.
from contextlib import contextmanager
from typing import Any
from fastlite import database
def get_db(db_path: str) -> Any:
"""
Get a database connection with required pragmas set.
Pragmas configured:
- journal_mode=WAL (write-ahead logging for better concurrency)
- synchronous=FULL (durability guarantee)
- foreign_keys=ON (enforce referential integrity)
- busy_timeout=5000 (wait up to 5s for locks)
Args:
db_path: Path to the SQLite database file.
Returns:
A fastlite database connection.
"""
db = database(db_path)
# Set pragmas for safety and performance
db.execute("PRAGMA journal_mode=WAL")
db.execute("PRAGMA synchronous=FULL")
db.execute("PRAGMA foreign_keys=ON")
db.execute("PRAGMA busy_timeout=5000")
return db
# Track which connections have an active transaction to prevent nesting
_active_transactions: set[int] = set()
@contextmanager
def transaction(db: Any):
"""
Context manager for BEGIN IMMEDIATE transactions.
Uses BEGIN IMMEDIATE to acquire a write lock immediately, preventing
concurrent write transactions from other connections.
Commits on successful completion, rolls back on exception.
Args:
db: A fastlite database connection.
Raises:
RuntimeError: If called while already in a transaction on this connection.
Example:
with transaction(db):
db.execute("INSERT INTO ...")
"""
conn_id = id(db)
if conn_id in _active_transactions:
raise RuntimeError("Nested transactions are not supported")
_active_transactions.add(conn_id)
try:
db.execute("BEGIN IMMEDIATE")
try:
yield
db.execute("COMMIT")
except Exception:
db.execute("ROLLBACK")
raise
finally:
_active_transactions.discard(conn_id)