1. Toni Ruža
  2. roktar

Commits

Toni Ruža  committed fe4b501

Basic update and delete methods for entities.

  • Participants
  • Parent commits c4df259
  • Branches default

Comments (0)

Files changed (4)

File roktar/backends/_interface.py

View file
  • Ignore whitespace
     @abstractmethod
     def get_all_entities(self):
         pass
+    
+    @abstractmethod
+    def update_entity(self, entity_id, data):
+        pass
+    
+    @abstractmethod
+    def delete_entity(self, entity_id):
+        pass

File roktar/backends/sqlite_backend.py

View file
  • Ignore whitespace
 
     def select_one(self, table_name, key_name, identity):
         return self.select(table_name, **{key_name: identity}).fetchone()
+    
+    def update(self, table_name, key_name, identity, **kwargs):
+        self.cursor.execute(
+            "UPDATE %s SET %s WHERE %s=?" % (
+                table_name,
+                ",".join(["%s=?" % k for k in kwargs.keys()]),
+                key_name
+            ), kwargs.values() + [identity]
+        )
+
+    def delete(self, table_name, **conditions):
+        self.db_connection.execute(
+            "DELETE FROM %s WHERE %s" % (
+                table_name,
+                " AND ".join("%s=?" % k for k in conditions.iterkeys())
+                ),
+            conditions.values()
+        )
 
 
 class SQLiteBackend(RoktarBackend, BrokerMixin):
         return entity_id
 
     def get_entity(self, entity_id):
-        data = self.select_one("Entities", "entity_id", entity_id)
-        return loads(str(data["cached_value"]))
+        try:
+            data = self.select_one("Entities", "entity_id", entity_id)
+            return loads(str(data["cached_value"]))
+        except TypeError:
+            raise errors.EntityNotFoundError()
 
     def get_changes(self, entity_id):
         return self.select("Changes", entity_id=entity_id)
             (data["entity_id"], loads(str(data["cached_value"])))
             for data in self.select("Entities")
         )
+
+    def update_entity(self, entity_id, data):
+        self.update("Entities", "entity_id", entity_id, cached_value=dumps(data))
+        self.commit()
+
+    def delete_entity(self, entity_id):
+        self.delete("Entities", entity_id=entity_id)
+        self.commit()

File roktar/errors.py

View file
  • Ignore whitespace
 
 class WrongFieldTypeError(Exception):
     pass
+
+class EntityNotFoundError(Exception):
+    pass

File tests/test_data_handling.py

View file
  • Ignore whitespace
 from roktar.configuration import Configuration
 from roktar.configuration_backends.json_backend import JSONConfiguration
 from roktar.backends.sqlite_backend import SQLiteBackend
+from roktar import errors
 
 
 class DataHandlingTests(unittest.TestCase):
         e_id, e = all_entities[0]
         self.assertEqual(e_id, entity_id)
         self.assertEqual(e, entity)
+        
+    def test_update_entity(self):
+        entity = dict(one=1)
+        entity_id = self.roktar.create_entity(entity)
+        entity["two"] = 2
+        self.roktar.update_entity(entity_id, entity)
+        self.assertEqual(entity, self.roktar.get_entity(entity_id))
+        
+    def test_delete_entity(self):
+        entity_id = self.roktar.create_entity(dict(hello="world"))
+        self.roktar.delete_entity(entity_id)
+        self.assertRaises(
+            errors.EntityNotFoundError,
+            lambda: self.roktar.get_entity(entity_id)
+        )