core/sqlite: extract immutable connection helper

use in bluemaestro/zotero modules

my/bluemaestro.py

@@ -10,7 +10,8 @@ import re
 import sqlite3
 from typing import Iterable, Sequence, Set, Optional
 
-from .core import get_files, LazyLogger, dataclass
+from my.core import get_files, LazyLogger, dataclass
+from my.core.sqlite import sqlite_connect_immutable
 
 from my.config import bluemaestro as config
 
@@ -50,8 +51,8 @@ def is_bad_table(name: str) -> bool:
     return False if delegate is None else delegate(name)
 
 
-from .core.cachew import cache_dir
-from .core.common import mcachew
+from my.core.cachew import cache_dir
+from my.core.common import mcachew
 @mcachew(depends_on=lambda: inputs(), cache_path=cache_dir('bluemaestro'))
 def measurements() -> Iterable[Measurement]:
     # todo ideally this would be via arguments... but needs to be lazy
@@ -66,7 +67,7 @@ def measurements() -> Iterable[Measurement]:
         tot = 0
         new = 0
         # todo assert increasing timestamp?
-        with sqlite3.connect(f'file:{f}?immutable=1', uri=True) as db:
+        with sqlite_connect_immutable(f) as db:
             db_dt: Optional[datetime] = None
             try:
                 datas = db.execute(f'SELECT "{f.name}" as name, Time, Temperature, Humidity, Pressure, Dewpoint FROM data ORDER BY log_index')
@@ -173,12 +174,12 @@ def measurements() -> Iterable[Measurement]:
     # for k, v in merged.items():
     #     yield Point(dt=k, temp=v) # meh?
 
-from .core import stat, Stats
+from my.core import stat, Stats
 def stats() -> Stats:
     return stat(measurements)
 
 
-from .core.pandas import DataFrameT, as_dataframe
+from my.core.pandas import DataFrameT, as_dataframe
 def dataframe() -> DataFrameT:
     """
     %matplotlib gtk
@@ -192,7 +193,7 @@ def dataframe() -> DataFrameT:
 
 
 def fill_influxdb() -> None:
-    from .core import influxdb
+    from my.core import influxdb
     influxdb.fill(measurements(), measurement=__name__)
 
 

my/core/dataset.py

@@ -1,12 +1,14 @@
-from pathlib import Path
+from .common import assert_subpackage; assert_subpackage(__name__)
+
+from .common import PathIsh
+from .sqlite import sqlite_connect_immutable
+
 
 # TODO wonder if also need to open without WAL.. test this on read-only directory/db file
-def connect_readonly(db: Path):
+def connect_readonly(db: PathIsh):
     import dataset # type: ignore
     # see https://github.com/pudo/dataset/issues/136#issuecomment-128693122
     # todo not sure if mode=ro has any benefit, but it doesn't work on read-only filesystems
     # maybe it should autodetect readonly filesystems and apply this? not sure
-    import sqlite3
-    # https://www.sqlite.org/draft/uri.html#uriimmutable
-    creator = lambda: sqlite3.connect(f'file:{db}?immutable=1', uri=True)
+    creator = lambda: sqlite_connect_immutable(db)
     return dataset.connect('sqlite:///', engine_kwargs={'creator': creator})

my/core/sqlite.py

@@ -0,0 +1,25 @@
+from .common import assert_subpackage; assert_subpackage(__name__)
+
+import sqlite3
+
+from .common import PathIsh
+
+def sqlite_connect_immutable(db: PathIsh) -> sqlite3.Connection:
+    # https://www.sqlite.org/draft/uri.html#uriimmutable
+    return sqlite3.connect(f'file:{db}?immutable=1', uri=True)
+
+
+from pathlib import Path
+def test_sqlite_connect_immutable(tmp_path: Path) -> None:
+    db = str(tmp_path / 'db.sqlite')
+    with sqlite3.connect(db) as conn:
+        conn.execute('CREATE TABLE testtable (col)')
+
+    import pytest  # type: ignore
+    with pytest.raises(sqlite3.OperationalError, match='readonly database'):
+        with sqlite_connect_immutable(db) as conn:
+            conn.execute('DROP TABLE testtable')
+
+    # succeeds without immutable
+    with sqlite3.connect(db) as conn:
+        conn.execute('DROP TABLE testtable')

tests/core.py

@@ -20,4 +20,5 @@ from my.core.freezer        import *
 from my.core.stats          import *
 from my.core.query          import *
 from my.core.serialize      import test_serialize_fallback
+from my.core.sqlite         import *
 from my.core.__main__       import *