delete-doccuments tests

dataloom/__init__.py

@@ -1,7 +1,15 @@
 from dataloom.loom import Dataloom
-from dataloom.exceptions import *
+from dataloom.exceptions import (
+    InvalidColumnValuesException,
+    InvalidFiltersForTableColumnException,
+    PkNotDefinedException,
+    TooManyPkException,
+    UnknownColumnException,
+    UnsupportedDialectException,
+    UnsupportedTypeException,
+)
 from dataloom.model import Model
-from dataloom.model.column import (
+from dataloom.model import (
     PrimaryKeyColumn,
     CreatedAtColumn,
     ForeignKeyColumn,

dataloom/model/__init__.py

@@ -1,7 +1,7 @@
 import inspect
 from dataloom.constants import CURRENT_TIME_STAMP, SQLITE_CURRENT_TIME_STAMP
 from dataloom.exceptions import UnknownColumnException, UnsupportedDialectException
-from dataloom.model.column import (
+from dataloom.columns import (
     Column,
     CreatedAtColumn,
     ForeignKeyColumn,

dataloom/statements/__init__.py

@@ -2,15 +2,14 @@
 import re
 from dataclasses import dataclass
 from typing import Optional
-
 from dataloom.exceptions import (
     InvalidColumnValuesException,
     InvalidFiltersForTableColumnException,
     PkNotDefinedException,
     TooManyPkException,
     UnsupportedDialectException,
 )
-from dataloom.model.column import (
+from dataloom.model import (
     Column,
     CreatedAtColumn,
     ForeignKeyColumn,

dataloom/tests/mysql/test_delete_mysql.py

@@ -0,0 +1,207 @@
+class TestDeletingOnMysql:
+    def test_delete_by_pk_fn(self):
+        from dataloom import (
+            Column,
+            PrimaryKeyColumn,
+            Dataloom,
+            TableColumn,
+            Model,
+            CreatedAtColumn,
+            UpdatedAtColumn,
+            ForeignKeyColumn,
+        )
+        from dataloom.keys import MySQLConfig
+        from typing import Optional
+
+        mysql_loom = Dataloom(
+            dialect="mysql",
+            database=MySQLConfig.database,
+            password=MySQLConfig.password,
+            user=MySQLConfig.user,
+        )
+
+        class User(Model):
+            __tablename__: Optional[TableColumn] = TableColumn(name="users")
+            id = PrimaryKeyColumn(type="int", auto_increment=True)
+            name = Column(type="text", nullable=False, default="Bob")
+            username = Column(type="varchar", unique=True, length=255)
+
+            # timestamps
+            createdAt = CreatedAtColumn()
+            updatedAt = UpdatedAtColumn()
+
+        class Post(Model):
+            __tablename__: Optional[TableColumn] = TableColumn(name="posts")
+            id = PrimaryKeyColumn(
+                type="int", auto_increment=True, nullable=False, unique=True
+            )
+            completed = Column(type="boolean", default=False)
+            title = Column(type="varchar", length=255, nullable=False)
+            # timestamps
+            createdAt = CreatedAtColumn()
+            updatedAt = UpdatedAtColumn()
+            # relations
+            userId = ForeignKeyColumn(
+                User, type="int", required=True, onDelete="CASCADE", onUpdate="CASCADE"
+            )
+
+        conn, _ = mysql_loom.connect_and_sync([Post, User], drop=True, force=True)
+
+        user = User(name="Crispen", username="heyy")
+        userId = mysql_loom.insert_one(user)
+        affected_rows_1 = mysql_loom.delete_by_pk(User, userId)
+        affected_rows_2 = mysql_loom.delete_by_pk(User, 89)
+        assert affected_rows_1 == 1
+        assert affected_rows_2 == 0
+        conn.close()
+
+    def test_delete_one_fn(self):
+        import pytest
+        from dataloom import (
+            Column,
+            PrimaryKeyColumn,
+            Dataloom,
+            TableColumn,
+            Model,
+            CreatedAtColumn,
+            UpdatedAtColumn,
+            ForeignKeyColumn,
+            UnknownColumnException,
+        )
+        from dataloom.keys import MySQLConfig
+        from typing import Optional
+
+        mysql_loom = Dataloom(
+            dialect="mysql",
+            database=MySQLConfig.database,
+            password=MySQLConfig.password,
+            user=MySQLConfig.user,
+        )
+
+        class User(Model):
+            __tablename__: Optional[TableColumn] = TableColumn(name="users")
+            id = PrimaryKeyColumn(type="int", auto_increment=True)
+            name = Column(type="text", nullable=False, default="Bob")
+            username = Column(type="varchar", unique=True, length=255)
+
+            # timestamps
+            createdAt = CreatedAtColumn()
+            updatedAt = UpdatedAtColumn()
+
+        class Post(Model):
+            __tablename__: Optional[TableColumn] = TableColumn(name="posts")
+            id = PrimaryKeyColumn(
+                type="int", auto_increment=True, nullable=False, unique=True
+            )
+            completed = Column(type="boolean", default=False)
+            title = Column(type="varchar", length=255, nullable=False)
+            # timestamps
+            createdAt = CreatedAtColumn()
+            updatedAt = UpdatedAtColumn()
+            # relations
+            userId = ForeignKeyColumn(
+                User, type="int", required=True, onDelete="CASCADE", onUpdate="CASCADE"
+            )
+
+        conn, _ = mysql_loom.connect_and_sync([Post, User], drop=True, force=True)
+        mysql_loom.insert_bulk(
+            [
+                User(name="Crispen", username="heyy"),
+                User(name="Crispen", username="who"),
+                User(name="Crispen", username="hi"),
+            ]
+        )
+        mysql_loom.delete_one(User, {"name": "Crispen"})
+        rows_1 = mysql_loom.find_many(User, {"name": "Crispen"})
+        mysql_loom.delete_one(User, {"name": "Crispen", "id": 9})
+        rows_2 = mysql_loom.find_many(User, {"name": "Crispen"})
+        mysql_loom.delete_one(User, {"name": "Crispen", "id": 2})
+        rows_3 = mysql_loom.find_many(User, {"name": "Crispen"})
+
+        with pytest.raises(UnknownColumnException) as exc_info:
+            mysql_loom.delete_bulk(User, {"location": "Crispen", "username": "@miller"})
+        assert str(exc_info.value) == "Table users does not have column 'location'."
+
+        assert len(rows_1) == 2
+        assert len(rows_2) == 2
+        assert len(rows_3) == 1
+        conn.close()
+
+    def test_delete_bulk_fn(self):
+        import pytest
+        from dataloom import (
+            Column,
+            PrimaryKeyColumn,
+            Dataloom,
+            TableColumn,
+            Model,
+            CreatedAtColumn,
+            UpdatedAtColumn,
+            ForeignKeyColumn,
+            UnknownColumnException,
+        )
+        from dataloom.keys import MySQLConfig
+        from typing import Optional
+
+        mysql_loom = Dataloom(
+            dialect="mysql",
+            database=MySQLConfig.database,
+            password=MySQLConfig.password,
+            user=MySQLConfig.user,
+        )
+
+        class User(Model):
+            __tablename__: Optional[TableColumn] = TableColumn(name="users")
+            id = PrimaryKeyColumn(type="int", auto_increment=True)
+            name = Column(type="text", nullable=False, default="Bob")
+            username = Column(type="varchar", unique=True, length=255)
+
+            # timestamps
+            createdAt = CreatedAtColumn()
+            updatedAt = UpdatedAtColumn()
+
+        class Post(Model):
+            __tablename__: Optional[TableColumn] = TableColumn(name="posts")
+            id = PrimaryKeyColumn(
+                type="int", auto_increment=True, nullable=False, unique=True
+            )
+            completed = Column(type="boolean", default=False)
+            title = Column(type="varchar", length=255, nullable=False)
+            # timestamps
+            createdAt = CreatedAtColumn()
+            updatedAt = UpdatedAtColumn()
+            # relations
+            userId = ForeignKeyColumn(
+                User, type="int", required=True, onDelete="CASCADE", onUpdate="CASCADE"
+            )
+
+        conn, _ = mysql_loom.connect_and_sync([Post, User], drop=True, force=True)
+        mysql_loom.insert_bulk(
+            [
+                User(name="Crispen", username="hi"),
+                User(name="Crispen", username="heyy"),
+                User(name="Crispen", username="hie"),
+            ]
+        )
+        mysql_loom.delete_bulk(User, {"name": "Crispen"})
+        rows_1 = mysql_loom.find_many(User, {"name": "Crispen"})
+        mysql_loom.insert_bulk(
+            [
+                User(name="Crispen", username="hi"),
+                User(name="Crispen", username="heyy"),
+                User(name="Crispen", username="hie"),
+            ]
+        )
+        mysql_loom.delete_bulk(User, {"name": "Crispen", "id": 99})
+        rows_2 = mysql_loom.find_many(User, {"name": "Crispen"})
+        mysql_loom.delete_bulk(User, {"name": "Crispen", "id": 5})
+        rows_3 = mysql_loom.find_many(User, {"name": "Crispen"})
+
+        with pytest.raises(UnknownColumnException) as exc_info:
+            mysql_loom.delete_bulk(User, {"location": "Crispen", "username": "@miller"})
+        assert str(exc_info.value) == "Table users does not have column 'location'."
+
+        assert len(rows_1) == 0
+        assert len(rows_2) == 3
+        assert len(rows_3) == 2
+        conn.close()

dataloom/tests/postgres/test_create_table_pg.py

@@ -22,7 +22,7 @@ class User(Model):
             name = Column(type="varchar", unique=True, length=255)
 
         with pytest.raises(Exception) as exc_info:
-            tables = pg_loom.sync([User], drop=True, force=True)
+            _ = pg_loom.sync([User], drop=True, force=True)
 
         assert (
             str(exc_info.value)
@@ -51,7 +51,7 @@ class User(Model):
             name = Column(type="varchar", unique=True, length=255)
 
         with pytest.raises(Exception) as exc_info:
-            tables = pg_loom.sync([User], drop=True, force=True)
+            _ = pg_loom.sync([User], drop=True, force=True)
         assert str(exc_info.value) == "Your table does not have a primary key column."
         conn.close()
 
@@ -107,14 +107,14 @@ class Post(Model):
             user=PgConfig.user,
             logging=True,
         )
-        conn, tables = pg_loom.connect_and_sync([User, Post], drop=True, force=True)
-        assert len(tables) == 2
+        conn, _ = pg_loom.connect_and_sync([User, Post], drop=True, force=True)
+        assert len(_) == 2
         assert conn.status == 1
-        assert sorted(tables) == sorted(["users", "posts"])
+        assert sorted(_) == sorted(["users", "posts"])
 
         conn.close()
 
-    def test_syncing_tables(self):
+    def test_syncing__(self):
         from dataloom import Model, Dataloom, Column, PrimaryKeyColumn, TableColumn
         from typing import Optional
         from dataloom.keys import PgConfig
@@ -140,7 +140,7 @@ class User(Model):
             username = Column(type="text", nullable=False, default="Hello there!!")
             name = Column(type="varchar", unique=True, length=255)
 
-        tables = pg_loom.sync([User, Post], drop=True, force=True)
-        assert len(tables) == 2
-        assert sorted(tables) == sorted(["users", "posts"])
+        _ = pg_loom.sync([User, Post], drop=True, force=True)
+        assert len(_) == 2
+        assert sorted(_) == sorted(["users", "posts"])
         conn.close()