Read-only mode

sqlitedict.py

@@ -106,6 +106,8 @@ def decode(obj):
 
 
 class SqliteDict(DictClass):
+    VALID_FLAGS = ['c', 'r', 'w', 'n']
+
     def __init__(self, filename=None, tablename='unnamed', flag='c',
                  autocommit=False, journal_mode="DELETE"):
         """
@@ -133,6 +135,11 @@ def __init__(self, filename=None, tablename='unnamed', flag='c',
         if self.in_temp:
             randpart = hex(random.randint(0, 0xffffff))[2:]
             filename = os.path.join(tempfile.gettempdir(), 'sqldict' + randpart)
+
+        if flag not in SqliteDict.VALID_FLAGS:
+            raise RuntimeError("Unrecognized flag: %s" % flag)
+        self.flag = flag
+
         if flag == 'n':
             if os.path.exists(filename):
                 os.remove(filename)
@@ -218,6 +225,9 @@ def __getitem__(self, key):
         return decode(item[0])
 
     def __setitem__(self, key, value):
+        if self.flag == 'r':
+            raise RuntimeError('Refusing to write to read-only SqliteDict')
+
         ADD_ITEM = 'REPLACE INTO %s (key, value) VALUES (?,?)' % self.tablename
         self.conn.execute(ADD_ITEM, (key, encode(value)))
 

tests/test_core.py

@@ -81,6 +81,30 @@ def test_overwrite_using_flag_n(self):
         next_db = sqlitedict.SqliteDict(filename=fname, tablename='sometable', flag='n')
         self.assertNotIn('key', next_db.keys())
 
+    def test_unrecognized_flag(self):
+
+        def build_with_bad_flag():
+            fname = norm_file('tests/db/sqlitedict-override-test.sqlite')
+            orig_db = sqlitedict.SqliteDict(filename=fname, flag = 'FOO')
+
+        with self.assertRaises(RuntimeError):
+            build_with_bad_flag()
+
+    def test_readonly(self):
+        fname = norm_file('tests/db/sqlitedict-override-test.sqlite')
+        orig_db = sqlitedict.SqliteDict(filename=fname)
+        orig_db['key'] = 'value'
+        orig_db.commit()
+        orig_db.close()
+
+        readonly_db = sqlitedict.SqliteDict(filename=fname, flag = 'r')
+        self.assertTrue(readonly_db['key'] == 'value')
+
+        def attempt_write():
+            readonly_db['key'] = ['new_value']
+
+        with self.assertRaises(RuntimeError):
+            attempt_write()
 
     def test_overwrite_using_flag_w(self):
         """Re-opening of a database with flag='w' destroys only the target table."""