Add pragmas to SQLite DBs and add retry logic to BaseModel

This commit is contained in:
2026-02-01 17:01:35 +01:00
parent 21b5a4c553
commit e37a9e84a6

View File

@@ -2,6 +2,7 @@
import logging
import os
import time
from . import datetime_naive_utc, datetime_aware_utc, now_utc
from .dns import encode_dnsname, EncodingError
@@ -78,11 +79,21 @@ class DateTimeFieldUTC(DateTimeField):
class BaseModel(Model):
"""Base model with database binding."""
"""Base model with database binding and save retry."""
class Meta:
database = db
def save(self, *args, max_retries=3, retry_delay=0.1, **kwargs):
"""Save with retry on DatabaseError (exponential backoff)."""
for attempt in range(max_retries):
try:
return super().save(*args, **kwargs)
except DatabaseError:
if attempt == max_retries - 1:
raise
time.sleep(retry_delay * (2 ** attempt))
class User(BaseModel):
"""User model for authentication."""
@@ -152,7 +163,10 @@ def init_database(config: dict):
db_dir = os.path.dirname(db_path)
if db_dir:
os.makedirs(db_dir, exist_ok=True)
actual_db = SqliteDatabase(db_path)
actual_db = SqliteDatabase(db_path, pragmas={
'journal_mode': 'wal',
'busy_timeout': 5000,
})
db.initialize(actual_db)
logging.debug(f"Database backend: SQLite path={db_path}")