fine-tune sqlite for production use by default

litequery/__init__.py

@@ -1,4 +1,4 @@
 from litequery.core import setup
 
-__version__ = "0.5.1"
+__version__ = "0.5.2"
 __all__ = ["setup"]

litequery/core.py

@@ -87,6 +87,15 @@ def dataclass_factory(cursor, row):
 
 
 class LitequeryBase:
+    PRAGMAS = [
+        ("journal_mode", "wal"),
+        ("foreign_keys", 1),
+        ("synchronous", "normal"),
+        ("mmap_size", 134217728),  # 128 Mb
+        ("journal_size_limit", 67108864),  # 64 Mb
+        ("cache_size", 2000),
+    ]
+
     def __init__(self, database, queries):
         self._database = database
         self._conn = None
@@ -126,6 +135,10 @@ async def query_method(**kwargs):
 
         return query_method
 
+    async def _apply_pragmas(self):
+        for pragma, value in self.PRAGMAS:
+            await self._conn.execute(f"pragma {pragma} = {value}")
+
     @asynccontextmanager
     async def transaction(self):
         conn = await self.get_connection()
@@ -143,7 +156,7 @@ async def transaction(self):
     async def connect(self):
         self._conn = await aiosqlite.connect(self._database)
         self._conn.row_factory = dataclass_factory
-        await self._conn.execute("PRAGMA foreign_keys = ON")
+        await self._apply_pragmas()
 
     async def disconnect(self):
         if self._conn is None:
@@ -180,6 +193,10 @@ def query_method(**kwargs):
 
         return query_method
 
+    def _apply_pragmas(self):
+        for pragma, value in self.PRAGMAS:
+            self._conn.execute(f"pragma {pragma} = {value}")
+
     @contextmanager
     def transaction(self):
         conn = self.get_connection()
@@ -197,7 +214,7 @@ def transaction(self):
     def connect(self):
         self._conn = sqlite3.connect(self._database)
         self._conn.row_factory = dataclass_factory
-        self._conn.execute("PRAGMA foreign_keys = ON")
+        self._apply_pragmas()
 
     def disconnect(self):
         if self._conn is None:

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "litequery"
-version = "0.5.1"
+version = "0.5.2"
 authors = [{ name = "Dima Charnyshou", email = "[email protected]" }]
 description = "A handy way to interact with an SQLite database from Python"
 readme = "README.md"

tests/test_core.py

@@ -222,3 +222,35 @@ def test_foreign_keys_enabled_sync(lq_sync):
 
     events = lq_sync.get_all_events()
     assert len(events) == 0
+
+
+@pytest.mark.asyncio
+async def test_pragmas_configured_async(lq_async):
+    expected_pragmas = [
+        ("journal_mode", "wal"),
+        ("foreign_keys", 1),
+        ("synchronous", 1),
+        ("mmap_size", 134217728),  # 128 Mb
+        ("journal_size_limit", 67108864),  # 64 Mb
+        ("cache_size", 2000),
+    ]
+    conn = await lq_async.get_connection()
+    for pragma, expected_value in expected_pragmas:
+        async with conn.execute(f"pragma {pragma}") as cursor:
+            result = await cursor.fetchone()
+            assert getattr(result, pragma) == expected_value
+
+
+def test_pragmas_configured_sync(lq_sync):
+    expected_pragmas = [
+        ("journal_mode", "wal"),
+        ("foreign_keys", 1),
+        ("synchronous", 1),
+        ("mmap_size", 134217728),  # 128 Mb
+        ("journal_size_limit", 67108864),  # 64 Mb
+        ("cache_size", 2000),
+    ]
+    conn = lq_sync.get_connection()
+    for pragma, expected_value in expected_pragmas:
+        result = conn.execute(f"pragma {pragma}").fetchone()
+        assert getattr(result, pragma) == expected_value