Merge pull request #1 from CrispenGari/delete

Delete

README.md

@@ -340,6 +340,26 @@ him = db.find_one(User, filters={"id": 1})
 print(him.to_dict())
 ```
 
+4. Deleting a record
+
+With the `delete_by_pk` method you can delete a record in a database based on the primary-key value:
+
+```py
+affected_rows = db.delete_by_pk(User, userId)
+```
+
+You cal also use `filters` to delete a record in a database. The `delete_one` function allows you to delete a single record in a database that matches a filter.
+
+```py
+affected_rows = db.delete_one(User, {"name": "Crispen"})
+```
+
+You can also the `delete_bulk` which delete a lot of records that matches a filter:
+
+```py
+affected_rows = db.delete_bulk(User, {"name": "Crispen"})
+```
+
 ### Associations
 
 With `dataloom` you can define models that have relationships. Let's say we have a model called `Post` and every post should belong to a single `User`. Here is how you can define model mappings between a `Post` and a `User` using the `ForeignKeyColumn()`

orm/db/__init__.py

@@ -10,7 +10,6 @@
     CreatedAtColumn,
     ForeignKeyColumn,
 )
-from functools import partial
 
 
 class Database:
@@ -49,6 +48,7 @@ def _execute_sql(
         fetchall=False,
         mutation=True,
         bulk: bool = False,
+        affected_rows: bool = False,
     ):
         # do we need to log the executed SQL?
         if self.logs:
@@ -62,7 +62,7 @@ def _execute_sql(
                     sql, vars=args
                 )
             # options
-            if bulk:
+            if bulk or affected_rows:
                 result = cursor.rowcount
             else:
                 if fetchmany:
@@ -163,7 +163,6 @@ def find_all(self, instance: Model):
         sql, _, __ = instance._get_select_where_stm(fields)
         data = list()
         rows = self._execute_sql(sql, fetchall=True)
-        print(len(rows))
         for row in rows:
             res = dict(zip(fields, row))
             data.append(instance(**res))
@@ -197,8 +196,6 @@ def find_by_pk(self, instance: Model, pk):
             elif isinstance(field, PrimaryKeyColumn):
                 pk_name = name
                 fields.append(name)
-
-        print(fields)
         sql, fields = instance._get_select_by_pk_stm(pk, pk_name, fields=fields)
         row = self._execute_sql(sql, fetchone=True)
         return None if row is None else instance(**dict(zip(fields, row)))
@@ -212,24 +209,31 @@ def find_one(self, instance: Model, filters: dict = {}):
         row = self._execute_sql(sql, args=params, fetchone=True)
         return None if row is None else instance(**dict(zip(fields, row)))
 
+    def delete_bulk(self, instance: Model, filters: dict = {}):
+        sql, params = instance._get_delete_bulk_where_stm(filters)
+        affected_rows = self._execute_sql(
+            sql, args=params, affected_rows=True, fetchall=True
+        )
+        return affected_rows
+
     def delete_one(self, instance: Model, filters: dict = {}):
-        fields = list()
+        pk = None
         for name, field in inspect.getmembers(instance):
-            if isinstance(field, Column):
-                fields.append(name)
-        sql, _, params = instance._get_select_where_stm(fields, filters)
-        row = self._execute_sql(sql, args=params, fetchone=True)
-        return None if row is None else instance(**dict(zip(fields, row)))
+            if isinstance(field, PrimaryKeyColumn):
+                pk = name
+        sql, params = instance._get_delete_where_stm(pk=pk, args=filters)
+        affected_rows = self._execute_sql(sql, args=params, affected_rows=True)
+        return affected_rows
 
     def delete_by_pk(self, instance: Model, pk):
         # what is the name of the primary key column?
         pk_name = "id"
-        fields = list()
         for name, field in inspect.getmembers(instance):
-            if isinstance(field, Column):
-                if field.primary_key:
-                    pk_name = name
-                fields.append(name)
-        sql, fields = instance._get_select_by_pk_stm(pk, pk_name, fields=fields)
-        row = self._execute_sql(sql, fetchone=True)
-        return None if row is None else instance(**dict(zip(fields, row)))
+            if isinstance(field, PrimaryKeyColumn):
+                pk_name = name
+
+        sql, pk = instance._get_delete_by_pk_stm(pk, pk_name)
+        affected_rows = self._execute_sql(
+            sql, args=(pk,), affected_rows=True, fetchall=True
+        )
+        return affected_rows

orm/keys.py

@@ -1,4 +1,4 @@
-push = True
+push = False
 
 
 if push:

orm/model/model.py

@@ -29,7 +29,11 @@ def _get_name(cls):
         for name, _ in inspect.getmembers(cls):
             if name == "__tablename__":
                 __tablename__ = cls.__tablename__
-        return cls.__name__.lower() if __tablename__ is None else __tablename__
+        return (
+            f'"{cls.__name__.lower()}"'
+            if __tablename__ is None
+            else f'"{__tablename__}"'
+        )
 
     @classmethod
     def _get_pk_attributes(cls):
@@ -190,6 +194,15 @@ def _get_select_by_pk_stm(cls, pk, pk_name: str = "id", fields: list = []):
         )
         return sql, fields
 
+    @classmethod
+    def _get_delete_by_pk_stm(cls, pk, pk_name: str = "id"):
+        sql = Statements.DELETE_BY_PK.format(
+            table_name=cls._get_name(),
+            pk="%s",  # mask it to avoid SQL Injection
+            pk_name=pk_name,
+        )
+        return sql, pk
+
     @classmethod
     def _get_insert_bulk_smt(cls, placeholders, columns, data):
         column_names = columns
@@ -249,3 +262,38 @@ def _get_select_where_stm(cls, fields: list = [], args: dict = {}):
                 filters=" AND ".join(filters),
             )
         return sql, fields, params
+
+    @classmethod
+    def _get_delete_where_stm(cls, pk: str = "id", args: dict = {}):
+        params = []
+        filters = []
+        for key, value in args.items():
+            filters.append(f"{key} = %s")
+            params.append(value)
+        if len(filters) == 0:
+            sql = Statements.DELETE_ALL_COMMAND.format(
+                table_name=cls._get_name(),
+            )
+        else:
+            sql = Statements.DELETE_ONE_WHERE_COMMAND.format(
+                table_name=cls._get_name(), filters=" AND ".join(filters), pk=pk
+            )
+        return sql, params
+
+    @classmethod
+    def _get_delete_bulk_where_stm(cls, args: dict = {}):
+        params = []
+        filters = []
+        for key, value in args.items():
+            filters.append(f"{key} = %s")
+            params.append(value)
+        if len(filters) == 0:
+            sql = Statements.DELETE_ALL_COMMAND.format(
+                table_name=cls._get_name(),
+            )
+        else:
+            sql = Statements.DELETE_BULK_WHERE_COMMAND.format(
+                table_name=cls._get_name(),
+                filters=" AND ".join(filters),
+            )
+        return sql, params

orm/model/statements.py

@@ -1,4 +1,13 @@
 class Statements:
+    # delete
+    DELETE_BY_PK = "DELETE FROM {table_name} WHERE {pk_name} = {pk};"
+    DELETE_ONE_WHERE_COMMAND = """
+    DELETE FROM {table_name} WHERE {pk} = (
+        SELECT {pk} FROM  {table_name} WHERE {filters} LIMIT 1
+    );
+    """
+    DELETE_BULK_WHERE_COMMAND = "DELETE FROM {table_name} WHERE {filters};"
+    DELETE_ALL_COMMAND = "DELETE FROM {table_name};"
     # select
     SELECT_COMMAND = "SELECT {column_names} FROM {table_name};"
     SELECT_BY_PK = "SELECT {column_names} FROM {table_name} WHERE {pk_name}={pk};"

orm/tests/test_create_table.py

@@ -9,14 +9,15 @@ def test_2_pk_error(self):
         db = Database(database, password=password, user=user)
         conn = db.connect()
 
-        class Users(Model):
+        class User(Model):
+            __tablename__ = "users"
             _id = PrimaryKeyColumn(type="bigint", auto_increment=True)
             id = PrimaryKeyColumn(type="bigint", auto_increment=True)
             username = Column(type="text", nullable=False, default="Hello there!!")
             name = Column(type="varchar", unique=True, length=255)
 
         with pytest.raises(Exception) as exc_info:
-            db.sync([Users], drop=True, force=True)
+            db.sync([User], drop=True, force=True)
 
         assert (
             str(exc_info.value)
@@ -34,12 +35,13 @@ def test_no_pk_error(self):
         db = Database(database, password=password, user=user)
         conn = db.connect()
 
-        class Users(Model):
+        class User(Model):
+            __tablename__ = "users"
             username = Column(type="text", nullable=False, default="Hello there!!")
             name = Column(type="varchar", unique=True, length=255)
 
         with pytest.raises(Exception) as exc_info:
-            db.sync([Users], drop=True, force=True)
+            db.sync([User], drop=True, force=True)
 
         assert str(exc_info.value) == "Your table does not have a primary key column."
         conn.close()
@@ -63,8 +65,8 @@ class User(Model):
             username = Column(type="text", nullable=False, default="Hello there!!")
             name = Column(type="varchar", unique=True, length=255)
 
-        assert User._get_name() == "users"
-        assert Todos._get_name() == "todos"
+        assert User._get_name() == '"users"'
+        assert Todos._get_name() == '"todos"'
         conn.close()
 
     def test_connect_sync(self):
@@ -90,7 +92,7 @@ class Post(Model):
 
         assert len(tables) == 2
         assert conn.status == 1
-        assert tables == ["users", "posts"]
+        assert sorted(tables) == sorted(["users", "posts"])
 
         conn.close()
 

orm/tests/test_delete.py

@@ -0,0 +1,102 @@
+class TestDeletingOnPG:
+    def test_delete_by_pk_single_fn(self):
+        from orm.db import Database
+        from orm.model.column import Column
+        from orm.model.model import Model, PrimaryKeyColumn
+        from orm.keys import password, database, user
+
+        db = Database(database, password=password, user=user)
+        conn = db.connect()
+
+        class User(Model):
+            __tablename__ = "users"
+            id = PrimaryKeyColumn(type="bigint", auto_increment=True)
+            username = Column(type="text", nullable=False, default="Hello there!!")
+            name = Column(type="varchar", unique=True, length=255)
+
+        db.sync([User], drop=True, force=True)
+
+        user = User(name="Crispen", username="heyy")
+        userId = db.commit(user)
+        affected_rows_1 = db.delete_by_pk(User, userId)
+        affected_rows_2 = db.delete_by_pk(User, 89)
+        assert affected_rows_1 == 1
+        assert affected_rows_2 == 0
+        conn.close()
+
+    def test_delete_one_fn(self):
+        from orm.db import Database
+        from orm.model.column import Column
+        from orm.model.model import Model, PrimaryKeyColumn
+        from orm.keys import password, database, user
+
+        db = Database(database, password=password, user=user)
+        conn = db.connect()
+
+        class User(Model):
+            __tablename__ = "users"
+            id = PrimaryKeyColumn(type="bigint", auto_increment=True)
+            username = Column(type="text", nullable=False, default="Hello there!!")
+            name = Column(type="varchar", unique=False, length=255)
+
+        db.sync([User], drop=True, force=True)
+
+        db.commit_bulk(
+            [
+                User(name="Crispen", username="heyy"),
+                User(name="Crispen", username="heyy"),
+                User(name="Crispen", username="heyy"),
+            ]
+        )
+        db.delete_one(User, {"name": "Crispen"})
+        rows_1 = db.find_many(User, {"name": "Crispen"})
+        db.delete_one(User, {"name": "Crispen", "id": 9})
+        rows_2 = db.find_many(User, {"name": "Crispen"})
+        db.delete_one(User, {"name": "Crispen", "id": 2})
+        rows_3 = db.find_many(User, {"name": "Crispen"})
+        assert len(rows_1) == 2
+        assert len(rows_2) == 2
+        assert len(rows_3) == 1
+        conn.close()
+
+    def test_delete_bulk_fn(self):
+        from orm.db import Database
+        from orm.model.column import Column
+        from orm.model.model import Model, PrimaryKeyColumn
+        from orm.keys import password, database, user
+
+        db = Database(database, password=password, user=user)
+        conn = db.connect()
+
+        class User(Model):
+            __tablename__ = "users"
+            id = PrimaryKeyColumn(type="bigint", auto_increment=True)
+            username = Column(type="text", nullable=False, default="Hello there!!")
+            name = Column(type="varchar", unique=False, length=255)
+
+        db.sync([User], drop=True, force=True)
+
+        db.commit_bulk(
+            [
+                User(name="Crispen", username="heyy"),
+                User(name="Crispen", username="heyy"),
+                User(name="Crispen", username="heyy"),
+            ]
+        )
+        db.delete_bulk(User, {"name": "Crispen"})
+        rows_1 = db.find_many(User, {"name": "Crispen"})
+        db.commit_bulk(
+            [
+                User(name="Crispen", username="heyy"),
+                User(name="Crispen", username="heyy"),
+                User(name="Crispen", username="heyy"),
+            ]
+        )
+        db.delete_bulk(User, {"name": "Crispen", "id": 99})
+        rows_2 = db.find_many(User, {"name": "Crispen"})
+        db.delete_bulk(User, {"name": "Crispen", "id": 5})
+        rows_3 = db.find_many(User, {"name": "Crispen"})
+        assert len(rows_1) == 0
+        assert len(rows_2) == 3
+        assert len(rows_3) == 2
+        conn.close()

orm/tests/test_insert.py

@@ -1,4 +1,4 @@
-class TestInsertingOnePG:
+class TestInsertingOnPG:
     def test_insetting_single_document(self):
         from orm.db import Database
         from orm.model.column import Column