From 4b6f4d1d0e63793fc91bd66fd29f2979f72c9ed8 Mon Sep 17 00:00:00 2001 From: Crispen Gari Date: Wed, 7 Feb 2024 09:32:06 +0200 Subject: [PATCH] insert-bulk with testing --- dataloom/loom/__init__.py | 33 +++- .../tests/mysql/test_create_tables_mysql.py | 23 +-- dataloom/tests/mysql/test_delete_mysql.py | 84 +++++++--- dataloom/tests/mysql/test_insert_mysql.py | 157 +++++++++++++++--- dataloom/tests/mysql/test_query_msql.py | 54 ++++-- dataloom/tests/mysql/test_update_mysql.py | 70 ++++---- .../tests/postgres/test_create_table_pg.py | 55 +++--- dataloom/tests/postgres/test_delete_pg.py | 67 ++++++-- dataloom/tests/postgres/test_insert_pg.py | 155 ++++++++++++++--- dataloom/tests/postgres/test_query_pg.py | 44 +++-- dataloom/tests/postgres/test_update_pg.py | 68 ++++---- dataloom/tests/sqlite3/test_delete_sqlite.py | 71 ++++++-- dataloom/tests/sqlite3/test_insert_sqlite.py | 154 ++++++++++++++--- dataloom/tests/sqlite3/test_query_sqlite.py | 52 ++++-- dataloom/tests/sqlite3/test_update_sqlite.py | 47 ++++-- dataloom/utils/__init__.py | 21 ++- hi.db | Bin 53248 -> 53248 bytes playground.py | 90 +++++----- 18 files changed, 900 insertions(+), 345 deletions(-) diff --git a/dataloom/loom/__init__.py b/dataloom/loom/__init__.py index f2c5d98..e005a99 100644 --- a/dataloom/loom/__init__.py +++ b/dataloom/loom/__init__.py @@ -281,9 +281,24 @@ def insert_one(self, instance: Model, values: ColumnValue | list[ColumnValue]): ) return row[0] if type(row) in [list, tuple] else row - def insert_bulk( - self, instance: Model, values: list[list[ColumnValue] | ColumnValue] - ): + def insert_bulk(self, instance: Model, values: list[list[ColumnValue]]): + # ? ensure that the values that are passed is a list of a list and they inner list have the same length + if not isinstance(values, list): + raise InvalidColumnValuesException( + "The insert_bulk method takes in values as lists of lists." + ) + all_list = [isinstance(v, list) for v in values] + if not all(all_list): + raise InvalidColumnValuesException( + "The insert_bulk method takes in values as lists of lists." + ) + lengths = [len(v) for v in values] + _max = max(lengths) + if not all([_max == v for v in lengths]): + raise InvalidColumnValuesException( + "The insert_bulk method takes in values as lists of lists with equal ColumnValues." + ) + columns = None placeholders = None data = [] @@ -296,13 +311,13 @@ def insert_bulk( if placeholders is None: placeholders = placeholder_values data.append(_values) - print(data) - # sql = instance._get_insert_bulk_smt( - # dialect=self.dialect, column_names=columns, placeholder_values=placeholders - # ) - # row_count = self._execute_sql(sql, args=tuple(data), fetchall=True, bulk=True) - # return row_count + sql = instance._get_insert_bulk_smt( + dialect=self.dialect, column_names=columns, placeholder_values=placeholders + ) + + row_count = self._execute_sql(sql, args=tuple(data), fetchall=True, bulk=True) + return row_count def find_many( self, diff --git a/dataloom/tests/mysql/test_create_tables_mysql.py b/dataloom/tests/mysql/test_create_tables_mysql.py index 0b92957..f3fefd1 100644 --- a/dataloom/tests/mysql/test_create_tables_mysql.py +++ b/dataloom/tests/mysql/test_create_tables_mysql.py @@ -2,7 +2,6 @@ class TestCreatingTableMysql: def test_2_pk_error(self): from dataloom import Column, PrimaryKeyColumn, Dataloom, TableColumn, Model import pytest - from typing import Optional from dataloom.keys import MySQLConfig mysql_loom = Dataloom( @@ -14,14 +13,14 @@ def test_2_pk_error(self): conn = mysql_loom.connect() class User(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="users") + __tablename__: TableColumn = TableColumn(name="users") _id = PrimaryKeyColumn(type="int", auto_increment=True) id = PrimaryKeyColumn(type="int", 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: - tables = mysql_loom.sync([User], drop=True, force=True) + _ = mysql_loom.sync([User], drop=True, force=True) assert ( str(exc_info.value) == "You have defined many field as primary keys which is not allowed. Fields (`_id`, `id`) are primary keys." @@ -31,7 +30,6 @@ class User(Model): def test_no_pk_error(self): import pytest from dataloom import Model, Dataloom, Column, TableColumn - from typing import Optional from dataloom.keys import MySQLConfig mysql_loom = Dataloom( @@ -43,18 +41,17 @@ def test_no_pk_error(self): conn = mysql_loom.connect() class User(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="users") + __tablename__: TableColumn = TableColumn(name="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: - tables = mysql_loom.sync([User], drop=True, force=True) + _ = mysql_loom.sync([User], drop=True, force=True) assert str(exc_info.value) == "Your table does not have a primary key column." conn.close() def test_table_name(self): from dataloom import Model, Dataloom, Column, PrimaryKeyColumn, TableColumn - from typing import Optional from dataloom.keys import MySQLConfig mysql_loom = Dataloom( @@ -71,7 +68,7 @@ class Posts(Model): title = Column(type="varchar", length=255, nullable=False) class User(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="users") + __tablename__: TableColumn = TableColumn(name="users") username = Column(type="text", nullable=False, default="Hello there!!") name = Column(type="varchar", unique=True, length=255) @@ -81,17 +78,16 @@ class User(Model): def test_connect_sync(self): from dataloom import Dataloom, Model, TableColumn, Column, PrimaryKeyColumn - from typing import Optional from dataloom.keys import MySQLConfig class User(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="users") + __tablename__: TableColumn = TableColumn(name="users") id = PrimaryKeyColumn(type="int", nullable=False, auto_increment=True) username = Column(type="text", nullable=False) name = Column(type="varchar", unique=False, length=255) class Post(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="posts") + __tablename__: TableColumn = TableColumn(name="posts") id = PrimaryKeyColumn(type="int", nullable=False, auto_increment=True) title = Column(type="text", nullable=False) @@ -110,7 +106,6 @@ class Post(Model): def test_syncing_tables(self): from dataloom import Model, Dataloom, Column, PrimaryKeyColumn, TableColumn - from typing import Optional from dataloom.keys import MySQLConfig mysql_loom = Dataloom( @@ -122,13 +117,13 @@ def test_syncing_tables(self): conn = mysql_loom.connect() class Post(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="posts") + __tablename__: TableColumn = TableColumn(name="posts") id = PrimaryKeyColumn(type="int", auto_increment=True) completed = Column(type="boolean", default=False) title = Column(type="varchar", length=255, nullable=False) class User(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="users") + __tablename__: TableColumn = TableColumn(name="users") id = PrimaryKeyColumn(type="int", auto_increment=True) username = Column(type="text", nullable=False, default="Hello there!!") name = Column(type="varchar", unique=True, length=255) diff --git a/dataloom/tests/mysql/test_delete_mysql.py b/dataloom/tests/mysql/test_delete_mysql.py index 8abd641..57f5afa 100644 --- a/dataloom/tests/mysql/test_delete_mysql.py +++ b/dataloom/tests/mysql/test_delete_mysql.py @@ -9,9 +9,9 @@ def test_delete_by_pk_fn(self): CreatedAtColumn, UpdatedAtColumn, ForeignKeyColumn, + ColumnValue, ) from dataloom.keys import MySQLConfig - from typing import Optional mysql_loom = Dataloom( dialect="mysql", @@ -21,7 +21,7 @@ def test_delete_by_pk_fn(self): ) class User(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="users") + __tablename__: 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) @@ -31,7 +31,7 @@ class User(Model): updatedAt = UpdatedAtColumn() class Post(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="posts") + __tablename__: TableColumn = TableColumn(name="posts") id = PrimaryKeyColumn( type="int", auto_increment=True, nullable=False, unique=True ) @@ -47,8 +47,10 @@ class Post(Model): conn, _ = mysql_loom.connect_and_sync([Post, User], drop=True, force=True) - user = User(name="Crispen", username="heyy") - userId = mysql_loom.insert_one(user) + userId = mysql_loom.insert_one( + User, ColumnValue(name="username", value="@miller") + ) + 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 @@ -68,9 +70,9 @@ def test_delete_one_fn(self): ForeignKeyColumn, UnknownColumnException, Filter, + ColumnValue, ) from dataloom.keys import MySQLConfig - from typing import Optional mysql_loom = Dataloom( dialect="mysql", @@ -80,7 +82,7 @@ def test_delete_one_fn(self): ) class User(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="users") + __tablename__: 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) @@ -90,7 +92,7 @@ class User(Model): updatedAt = UpdatedAtColumn() class Post(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="posts") + __tablename__: TableColumn = TableColumn(name="posts") id = PrimaryKeyColumn( type="int", auto_increment=True, nullable=False, unique=True ) @@ -106,11 +108,21 @@ class Post(Model): 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"), - ] + User, + values=[ + [ + ColumnValue(name="name", value="Crispen"), + ColumnValue(name="username", value="hi"), + ], + [ + ColumnValue(name="name", value="Crispen"), + ColumnValue(name="username", value="heyy"), + ], + [ + ColumnValue(name="name", value="Crispen"), + ColumnValue(name="username", value="who"), + ], + ], ) mysql_loom.delete_one(User, filters=[Filter(column="name", value="Crispen")]) rows_1 = mysql_loom.find_many( @@ -165,9 +177,9 @@ def test_delete_bulk_fn(self): ForeignKeyColumn, UnknownColumnException, Filter, + ColumnValue, ) from dataloom.keys import MySQLConfig - from typing import Optional mysql_loom = Dataloom( dialect="mysql", @@ -177,7 +189,7 @@ def test_delete_bulk_fn(self): ) class User(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="users") + __tablename__: 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) @@ -187,7 +199,7 @@ class User(Model): updatedAt = UpdatedAtColumn() class Post(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="posts") + __tablename__: TableColumn = TableColumn(name="posts") id = PrimaryKeyColumn( type="int", auto_increment=True, nullable=False, unique=True ) @@ -203,11 +215,21 @@ class Post(Model): 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"), - ] + User, + values=[ + [ + ColumnValue(name="name", value="Crispen"), + ColumnValue(name="username", value="hi"), + ], + [ + ColumnValue(name="name", value="Crispen"), + ColumnValue(name="username", value="heyy"), + ], + [ + ColumnValue(name="name", value="Crispen"), + ColumnValue(name="username", value="hie"), + ], + ], ) mysql_loom.delete_bulk( User, filters=Filter(column="name", value="Crispen", operator="eq") @@ -216,11 +238,21 @@ class Post(Model): User, filters=Filter(column="name", value="Crispen", operator="eq") ) mysql_loom.insert_bulk( - [ - User(name="Crispen", username="hi"), - User(name="Crispen", username="heyy"), - User(name="Crispen", username="hie"), - ] + User, + values=[ + [ + ColumnValue(name="name", value="Crispen"), + ColumnValue(name="username", value="hi"), + ], + [ + ColumnValue(name="name", value="Crispen"), + ColumnValue(name="username", value="heyy"), + ], + [ + ColumnValue(name="name", value="Crispen"), + ColumnValue(name="username", value="hie"), + ], + ], ) mysql_loom.delete_bulk( User, diff --git a/dataloom/tests/mysql/test_insert_mysql.py b/dataloom/tests/mysql/test_insert_mysql.py index 2e95e6e..19533ed 100644 --- a/dataloom/tests/mysql/test_insert_mysql.py +++ b/dataloom/tests/mysql/test_insert_mysql.py @@ -9,9 +9,9 @@ def test_insetting_single_document(self): UpdatedAtColumn, TableColumn, ForeignKeyColumn, + ColumnValue, ) from dataloom.keys import MySQLConfig - from typing import Optional mysql_loom = Dataloom( dialect="mysql", @@ -21,7 +21,7 @@ def test_insetting_single_document(self): ) class User(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="users") + __tablename__: 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) @@ -31,7 +31,7 @@ class User(Model): updatedAt = UpdatedAtColumn() class Post(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="posts") + __tablename__: TableColumn = TableColumn(name="posts") id = PrimaryKeyColumn( type="int", auto_increment=True, nullable=False, unique=True ) @@ -46,10 +46,16 @@ class Post(Model): ) conn, _ = mysql_loom.connect_and_sync([Post, User], drop=True, force=True) - user = User(username="@miller") - userId = mysql_loom.insert_one(user) - post = Post(title="What are you doing?", userId=userId) - post_id = mysql_loom.insert_one(post) + userId = mysql_loom.insert_one( + User, ColumnValue(name="username", value="@miller") + ) + post_id = mysql_loom.insert_one( + Post, + values=[ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ], + ) assert userId == 1 assert post_id == 1 conn.close() @@ -64,9 +70,9 @@ def test_insetting_multiple_document(self): UpdatedAtColumn, TableColumn, ForeignKeyColumn, + ColumnValue, ) from dataloom.keys import MySQLConfig - from typing import Optional mysql_loom = Dataloom( dialect="mysql", @@ -76,7 +82,7 @@ def test_insetting_multiple_document(self): ) class User(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="users") + __tablename__: 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) @@ -86,7 +92,7 @@ class User(Model): updatedAt = UpdatedAtColumn() class Post(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="posts") + __tablename__: TableColumn = TableColumn(name="posts") id = PrimaryKeyColumn( type="int", auto_increment=True, nullable=False, unique=True ) @@ -101,10 +107,15 @@ class Post(Model): ) conn, _ = mysql_loom.connect_and_sync([Post, User], drop=True, force=True) - user = User(username="@miller") - userId = mysql_loom.insert_one(user) - post = Post(title="What are you doing?", userId=userId) - count = mysql_loom.insert_bulk([post for i in range(5)]) + userId = mysql_loom.insert_one( + User, values=ColumnValue(name="username", value="@miller") + ) + values = [ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ] + + count = mysql_loom.insert_bulk(Post, values=[values for i in range(5)]) assert count == 5 conn.close() @@ -117,6 +128,8 @@ def test_relational_instances(self): PrimaryKeyColumn, Column, Dataloom, + ColumnValue, + TableColumn, ) from dataloom.keys import MySQLConfig @@ -128,7 +141,7 @@ def test_relational_instances(self): ) class User(Model): - __tablename__ = "users" + __tablename__: TableColumn = TableColumn(name="users") id = PrimaryKeyColumn(type="bigint", auto_increment=True) username = Column(type="text", nullable=False) name = Column(type="varchar", unique=False, length=255) @@ -136,7 +149,7 @@ class User(Model): updatedAt = UpdatedAtColumn() class Post(Model): - __tablename__ = "posts" + __tablename__: TableColumn = TableColumn(name="posts") id = PrimaryKeyColumn(type="bigint", auto_increment=True) title = Column(type="text", nullable=False, default="Hello there!!") @@ -147,11 +160,115 @@ class Post(Model): ) conn, _ = mysql_loom.connect_and_sync([User, Post], drop=True, force=True) - user = User(name="Crispen", username="heyy") - userId = mysql_loom.insert_one(user) - postId = mysql_loom.insert_one( - Post(userId=userId, title="What are you thinking"), + + userId = mysql_loom.insert_one( + User, ColumnValue(name="username", value="@miller") ) + values = [ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ] + + postId = mysql_loom.insert_one(Post, values) now = mysql_loom.find_by_pk(Post, postId) assert userId == now["userId"] conn.close() + + def test_insert_bulk_with_errors(self): + from dataloom import ( + Dataloom, + Model, + Column, + PrimaryKeyColumn, + CreatedAtColumn, + UpdatedAtColumn, + TableColumn, + ForeignKeyColumn, + ColumnValue, + InvalidColumnValuesException, + ) + from dataloom.keys import MySQLConfig + import pytest + + mysql_loom = Dataloom( + dialect="mysql", + database=MySQLConfig.database, + password=MySQLConfig.password, + user=MySQLConfig.user, + ) + + class User(Model): + __tablename__: 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__: 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) + userId = mysql_loom.insert_one( + User, values=ColumnValue(name="username", value="@miller") + ) + values = [ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ] + with pytest.raises(InvalidColumnValuesException) as exc_info: + mysql_loom.insert_bulk(Post, values=values) + + assert ( + str(exc_info.value) + == "The insert_bulk method takes in values as lists of lists." + ) + + with pytest.raises(InvalidColumnValuesException) as exc_info: + mysql_loom.insert_bulk( + Post, + values=[ + [ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ], + ColumnValue(name="title", value="What are you doing?"), + ], + ) + assert ( + str(exc_info.value) + == "The insert_bulk method takes in values as lists of lists." + ) + + with pytest.raises(InvalidColumnValuesException) as exc_info: + mysql_loom.insert_bulk( + Post, + values=[ + [ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ], + [ + ColumnValue(name="title", value="What are you doing?"), + ], + ], + ) + assert ( + str(exc_info.value) + == "The insert_bulk method takes in values as lists of lists with equal ColumnValues." + ) + conn.close() diff --git a/dataloom/tests/mysql/test_query_msql.py b/dataloom/tests/mysql/test_query_msql.py index 66e2b66..2d02819 100644 --- a/dataloom/tests/mysql/test_query_msql.py +++ b/dataloom/tests/mysql/test_query_msql.py @@ -10,6 +10,7 @@ def test_find_by_pk_fn(self): TableColumn, ForeignKeyColumn, UnknownColumnException, + ColumnValue, ) import pytest from dataloom.keys import MySQLConfig @@ -44,10 +45,15 @@ class Post(Model): ) conn, _ = mysql_loom.connect_and_sync([Post, User], drop=True, force=True) - user = User(username="@miller") - userId = mysql_loom.insert_one(user) - post = Post(title="What are you doing?", userId=userId) - _ = mysql_loom.insert_bulk([post for i in range(5)]) + + userId = mysql_loom.insert_one( + User, ColumnValue(name="username", value="@miller") + ) + values = [ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ] + _ = mysql_loom.insert_bulk(Post, [values for i in range(5)]) me = mysql_loom.find_by_pk(User, 1) her = mysql_loom.find_by_pk(User, 2) @@ -76,6 +82,7 @@ def test_find_all_fn(self): TableColumn, ForeignKeyColumn, UnknownColumnException, + ColumnValue, ) from dataloom.keys import MySQLConfig from typing import Optional @@ -110,10 +117,14 @@ class Post(Model): ) conn, _ = mysql_loom.connect_and_sync([Post, User], drop=True, force=True) - user = User(username="@miller") - userId = mysql_loom.insert_one(user) - post = Post(title="What are you doing?", userId=userId) - _ = mysql_loom.insert_bulk([post for i in range(5)]) + userId = mysql_loom.insert_one( + User, ColumnValue(name="username", value="@miller") + ) + values = [ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ] + _ = mysql_loom.insert_bulk(Post, [values for i in range(5)]) users = mysql_loom.find_all(User) posts = mysql_loom.find_all(Post) paginated = mysql_loom.find_all( @@ -149,6 +160,7 @@ def test_find_one_fn(self): ForeignKeyColumn, UnknownColumnException, Filter, + ColumnValue, ) from dataloom.keys import MySQLConfig from typing import Optional @@ -183,10 +195,14 @@ class Post(Model): ) conn, _ = mysql_loom.connect_and_sync([Post, User], drop=True, force=True) - user = User(username="@miller") - userId = mysql_loom.insert_one(user) - post = Post(title="What are you doing?", userId=userId) - _ = mysql_loom.insert_bulk([post for i in range(5)]) + userId = mysql_loom.insert_one( + User, ColumnValue(name="username", value="@miller") + ) + values = [ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ] + _ = mysql_loom.insert_bulk(Post, [values for i in range(5)]) one_0 = mysql_loom.find_one(User, filters=Filter(column="id", value=5)) one_1 = mysql_loom.find_one(User, filters=Filter(column="id", value=1)) @@ -249,6 +265,7 @@ def test_find_many(self): ForeignKeyColumn, UnknownColumnException, Filter, + ColumnValue, ) from dataloom.keys import MySQLConfig from typing import Optional @@ -283,10 +300,15 @@ class Post(Model): ) conn, _ = mysql_loom.connect_and_sync([Post, User], drop=True, force=True) - user = User(username="@miller") - userId = mysql_loom.insert_one(user) - post = Post(title="What are you doing?", userId=userId) - rows = mysql_loom.insert_bulk([post for i in range(5)]) + userId = mysql_loom.insert_one( + User, ColumnValue(name="username", value="@miller") + ) + values = [ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ] + rows = mysql_loom.insert_bulk(Post, [values for i in range(5)]) + posts = mysql_loom.find_many( Post, filters=[Filter(column="id", value=1), Filter(column="userId", value=1)], diff --git a/dataloom/tests/mysql/test_update_mysql.py b/dataloom/tests/mysql/test_update_mysql.py index 98698fd..e54c8e1 100644 --- a/dataloom/tests/mysql/test_update_mysql.py +++ b/dataloom/tests/mysql/test_update_mysql.py @@ -1,7 +1,6 @@ class TestUpdateOnMySQL: def test_update_by_pk_fn(self): import time - from typing import Optional from dataloom import ( Column, @@ -24,7 +23,7 @@ def test_update_by_pk_fn(self): ) class User(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="users") + __tablename__: 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) @@ -34,7 +33,7 @@ class User(Model): updatedAt = UpdatedAtColumn() class Post(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="posts") + __tablename__: TableColumn = TableColumn(name="posts") id = PrimaryKeyColumn( type="int", auto_increment=True, nullable=False, unique=True ) @@ -50,10 +49,15 @@ class Post(Model): conn, _ = mysql_loom.connect_and_sync([Post, User], drop=True, force=True) - user = User(username="@miller") - userId = mysql_loom.insert_one(user) - post = Post(title="What are you doing?", userId=userId) - _ = mysql_loom.insert_bulk([post for i in range(5)]) + userId = mysql_loom.insert_one( + User, values=ColumnValue(name="username", value="@miller") + ) + post = [ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ] + + _ = mysql_loom.insert_bulk(Post, [post for i in range(5)]) time.sleep(0.05) res_1 = mysql_loom.update_by_pk( User, userId, ColumnValue(name="username", value="Gari") @@ -70,7 +74,6 @@ class Post(Model): def test_update_one_fn(self): import time - from typing import Optional import pytest @@ -97,7 +100,7 @@ def test_update_one_fn(self): ) class User(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="users") + __tablename__: 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) @@ -107,7 +110,7 @@ class User(Model): updatedAt = UpdatedAtColumn() class Post(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="posts") + __tablename__: TableColumn = TableColumn(name="posts") id = PrimaryKeyColumn( type="int", auto_increment=True, nullable=False, unique=True ) @@ -122,10 +125,15 @@ class Post(Model): conn, _ = mysql_loom.connect_and_sync([Post, User], drop=True, force=True) - user = User(username="@miller") - userId = mysql_loom.insert_one(user) - post = Post(title="What are you doing?", userId=userId) - _ = mysql_loom.insert_bulk([post for i in range(5)]) + userId = mysql_loom.insert_one( + User, values=ColumnValue(name="username", value="@miller") + ) + post = [ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ] + + _ = mysql_loom.insert_bulk(Post, [post for i in range(5)]) time.sleep(0.05) res_1 = mysql_loom.update_one( User, @@ -160,7 +168,6 @@ class Post(Model): conn.close() def test_update_bulk_fn(self): - from typing import Optional import pytest from dataloom import ( Column, @@ -185,7 +192,7 @@ def test_update_bulk_fn(self): ) class User(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="users") + __tablename__: 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) @@ -195,7 +202,7 @@ class User(Model): updatedAt = UpdatedAtColumn() class Post(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="posts") + __tablename__: TableColumn = TableColumn(name="posts") id = PrimaryKeyColumn( type="int", auto_increment=True, nullable=False, unique=True ) @@ -210,10 +217,15 @@ class Post(Model): conn, _ = mysql_loom.connect_and_sync([Post, User], drop=True, force=True) - user = User(username="@miller") - userId = mysql_loom.insert_one(user) - post = Post(title="What are you doing?", userId=userId) - _ = mysql_loom.insert_bulk([post for i in range(5)]) + userId = mysql_loom.insert_one( + User, values=ColumnValue(name="username", value="@miller") + ) + post = [ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ] + + _ = mysql_loom.insert_bulk(Post, [post for i in range(5)]) res_1 = mysql_loom.update_bulk( Post, Filter(column="userId", value=userId), @@ -255,7 +267,6 @@ class Post(Model): conn.close() def test_increment_fn(self): - from typing import Optional import pytest from dataloom import ( Column, @@ -278,7 +289,7 @@ def test_increment_fn(self): ) class User(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="users") + __tablename__: 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) @@ -289,8 +300,9 @@ class User(Model): conn, _ = mysql_loom.connect_and_sync([User], drop=True, force=True) - user = User(username="@miller") - userId = mysql_loom.insert_one(user) + userId = mysql_loom.insert_one( + User, values=ColumnValue(name="username", value="@miller") + ) affected_rows = mysql_loom.increment( User, @@ -322,7 +334,6 @@ class User(Model): conn.close() def test_decrement_fn(self): - from typing import Optional import pytest from dataloom import ( Column, @@ -345,7 +356,7 @@ def test_decrement_fn(self): ) class User(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="users") + __tablename__: 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) @@ -356,8 +367,9 @@ class User(Model): conn, _ = mysql_loom.connect_and_sync([User], drop=True, force=True) - user = User(username="@miller") - userId = mysql_loom.insert_one(user) + userId = mysql_loom.insert_one( + User, values=ColumnValue(name="username", value="@miller") + ) affected_rows = mysql_loom.decrement( User, diff --git a/dataloom/tests/postgres/test_create_table_pg.py b/dataloom/tests/postgres/test_create_table_pg.py index d22cdb3..d6ac5c9 100644 --- a/dataloom/tests/postgres/test_create_table_pg.py +++ b/dataloom/tests/postgres/test_create_table_pg.py @@ -1,9 +1,8 @@ class TestCreatingTablePG: def test_2_pk_error(self): from dataloom import Column, PrimaryKeyColumn, Dataloom, TableColumn, Model - from dataloom.keys import PgConfig import pytest - from typing import Optional + from dataloom.keys import PgConfig pg_loom = Dataloom( dialect="postgres", @@ -14,15 +13,14 @@ def test_2_pk_error(self): conn = pg_loom.connect() class User(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="users") - _id = PrimaryKeyColumn(type="bigint", auto_increment=True) - id = PrimaryKeyColumn(type="bigint", auto_increment=True) + __tablename__: TableColumn = TableColumn(name="users") + _id = PrimaryKeyColumn(type="int", auto_increment=True) + id = PrimaryKeyColumn(type="int", 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: _ = pg_loom.sync([User], drop=True, force=True) - assert ( str(exc_info.value) == 'You have defined many field as primary keys which is not allowed. Fields ("_id", "id") are primary keys.' @@ -31,9 +29,8 @@ class User(Model): def test_no_pk_error(self): import pytest - from dataloom.keys import PgConfig from dataloom import Model, Dataloom, Column, TableColumn - from typing import Optional + from dataloom.keys import PgConfig pg_loom = Dataloom( dialect="postgres", @@ -44,7 +41,7 @@ def test_no_pk_error(self): conn = pg_loom.connect() class User(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="users") + __tablename__: TableColumn = TableColumn(name="users") username = Column(type="text", nullable=False, default="Hello there!!") name = Column(type="varchar", unique=True, length=255) @@ -55,7 +52,6 @@ class User(Model): def test_table_name(self): from dataloom import Model, Dataloom, Column, PrimaryKeyColumn, TableColumn - from typing import Optional from dataloom.keys import PgConfig pg_loom = Dataloom( @@ -67,12 +63,12 @@ def test_table_name(self): conn = pg_loom.connect() class Posts(Model): - id = PrimaryKeyColumn(type="bigint", auto_increment=True) + id = PrimaryKeyColumn(type="int", auto_increment=True) completed = Column(type="boolean", default=False) title = Column(type="varchar", length=255, nullable=False) class User(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="users") + __tablename__: TableColumn = TableColumn(name="users") username = Column(type="text", nullable=False, default="Hello there!!") name = Column(type="varchar", unique=True, length=255) @@ -82,20 +78,19 @@ class User(Model): def test_connect_sync(self): from dataloom import Dataloom, Model, TableColumn, Column, PrimaryKeyColumn - from typing import Optional from dataloom.keys import PgConfig class User(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="users") - id = PrimaryKeyColumn(type="bigint", nullable=False, auto_increment=True) + __tablename__: TableColumn = TableColumn(name="users") + id = PrimaryKeyColumn(type="int", nullable=False, auto_increment=True) username = Column(type="text", nullable=False) name = Column(type="varchar", unique=False, length=255) class Post(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="posts") + __tablename__: TableColumn = TableColumn(name="posts") - id = PrimaryKeyColumn(type="bigint", nullable=False, auto_increment=True) - title = Column(type="text", nullable=False, default="Hello there!!") + id = PrimaryKeyColumn(type="int", nullable=False, auto_increment=True) + title = Column(type="text", nullable=False) pg_loom = Dataloom( dialect="postgres", @@ -103,14 +98,14 @@ class Post(Model): password=PgConfig.password, user=PgConfig.user, ) - conn, _ = pg_loom.connect_and_sync([User, Post], drop=True, force=True) - assert len(_) >= 2 - assert "users" in _ and "posts" in _ + conn, tables = pg_loom.connect_and_sync([User, Post], drop=True, force=True) + assert len(tables) >= 2 + assert "users" in tables and "posts" in tables + conn.close() - def test_syncing__(self): + def test_syncing_tables(self): from dataloom import Model, Dataloom, Column, PrimaryKeyColumn, TableColumn - from typing import Optional from dataloom.keys import PgConfig pg_loom = Dataloom( @@ -122,18 +117,18 @@ def test_syncing__(self): conn = pg_loom.connect() class Post(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="posts") - id = PrimaryKeyColumn(type="bigint", auto_increment=True) + __tablename__: TableColumn = TableColumn(name="posts") + id = PrimaryKeyColumn(type="int", auto_increment=True) completed = Column(type="boolean", default=False) title = Column(type="varchar", length=255, nullable=False) class User(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="users") - id = PrimaryKeyColumn(type="bigint", auto_increment=True) + __tablename__: TableColumn = TableColumn(name="users") + id = PrimaryKeyColumn(type="int", auto_increment=True) username = Column(type="text", nullable=False, default="Hello there!!") name = Column(type="varchar", unique=True, length=255) - _ = pg_loom.sync([User, Post], drop=True, force=True) - assert len(_) >= 2 - assert "users" in _ and "posts" in _ + tables = pg_loom.sync([User, Post], drop=True, force=True) + assert len(tables) >= 2 + assert "users" in tables and "posts" in tables conn.close() diff --git a/dataloom/tests/postgres/test_delete_pg.py b/dataloom/tests/postgres/test_delete_pg.py index 12a821d..bda8e98 100644 --- a/dataloom/tests/postgres/test_delete_pg.py +++ b/dataloom/tests/postgres/test_delete_pg.py @@ -9,6 +9,7 @@ def test_delete_by_pk_fn(self): CreatedAtColumn, UpdatedAtColumn, ForeignKeyColumn, + ColumnValue, ) from dataloom.keys import PgConfig from typing import Optional @@ -47,8 +48,8 @@ class Post(Model): conn, _ = pg_loom.connect_and_sync([Post, User], drop=True, force=True) - user = User(name="Crispen", username="heyy") - userId = pg_loom.insert_one(user) + userId = pg_loom.insert_one(User, ColumnValue(name="username", value="@miller")) + affected_rows_1 = pg_loom.delete_by_pk(User, userId) affected_rows_2 = pg_loom.delete_by_pk(User, 89) assert affected_rows_1 == 1 @@ -68,6 +69,7 @@ def test_delete_one_fn(self): ForeignKeyColumn, UnknownColumnException, Filter, + ColumnValue, ) from dataloom.keys import PgConfig from typing import Optional @@ -106,11 +108,21 @@ class Post(Model): conn, _ = pg_loom.connect_and_sync([Post, User], drop=True, force=True) pg_loom.insert_bulk( - [ - User(name="Crispen", username="heyy"), - User(name="Crispen", username="who"), - User(name="Crispen", username="hi"), - ] + User, + values=[ + [ + ColumnValue(name="name", value="Crispen"), + ColumnValue(name="username", value="hi"), + ], + [ + ColumnValue(name="name", value="Crispen"), + ColumnValue(name="username", value="heyy"), + ], + [ + ColumnValue(name="name", value="Crispen"), + ColumnValue(name="username", value="who"), + ], + ], ) pg_loom.delete_one(User, filters=[Filter(column="name", value="Crispen")]) rows_1 = pg_loom.find_many( @@ -165,6 +177,7 @@ def test_delete_bulk_fn(self): ForeignKeyColumn, UnknownColumnException, Filter, + ColumnValue, ) from dataloom.keys import PgConfig from typing import Optional @@ -203,11 +216,21 @@ class Post(Model): conn, _ = pg_loom.connect_and_sync([Post, User], drop=True, force=True) pg_loom.insert_bulk( - [ - User(name="Crispen", username="hi"), - User(name="Crispen", username="heyy"), - User(name="Crispen", username="hie"), - ] + User, + values=[ + [ + ColumnValue(name="name", value="Crispen"), + ColumnValue(name="username", value="hi"), + ], + [ + ColumnValue(name="name", value="Crispen"), + ColumnValue(name="username", value="heyy"), + ], + [ + ColumnValue(name="name", value="Crispen"), + ColumnValue(name="username", value="hie"), + ], + ], ) pg_loom.delete_bulk( User, filters=Filter(column="name", value="Crispen", operator="eq") @@ -216,11 +239,21 @@ class Post(Model): User, filters=Filter(column="name", value="Crispen", operator="eq") ) pg_loom.insert_bulk( - [ - User(name="Crispen", username="hi"), - User(name="Crispen", username="heyy"), - User(name="Crispen", username="hie"), - ] + User, + values=[ + [ + ColumnValue(name="name", value="Crispen"), + ColumnValue(name="username", value="hi"), + ], + [ + ColumnValue(name="name", value="Crispen"), + ColumnValue(name="username", value="heyy"), + ], + [ + ColumnValue(name="name", value="Crispen"), + ColumnValue(name="username", value="hie"), + ], + ], ) pg_loom.delete_bulk( User, diff --git a/dataloom/tests/postgres/test_insert_pg.py b/dataloom/tests/postgres/test_insert_pg.py index 694621d..078a1be 100644 --- a/dataloom/tests/postgres/test_insert_pg.py +++ b/dataloom/tests/postgres/test_insert_pg.py @@ -9,9 +9,9 @@ def test_insetting_single_document(self): UpdatedAtColumn, TableColumn, ForeignKeyColumn, + ColumnValue, ) from dataloom.keys import PgConfig - from typing import Optional pg_loom = Dataloom( dialect="postgres", @@ -21,7 +21,7 @@ def test_insetting_single_document(self): ) class User(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="users") + __tablename__: 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) @@ -31,7 +31,7 @@ class User(Model): updatedAt = UpdatedAtColumn() class Post(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="posts") + __tablename__: TableColumn = TableColumn(name="posts") id = PrimaryKeyColumn( type="int", auto_increment=True, nullable=False, unique=True ) @@ -46,10 +46,14 @@ class Post(Model): ) conn, _ = pg_loom.connect_and_sync([Post, User], drop=True, force=True) - user = User(username="@miller") - userId = pg_loom.insert_one(user) - post = Post(title="What are you doing?", userId=userId) - post_id = pg_loom.insert_one(post) + userId = pg_loom.insert_one(User, ColumnValue(name="username", value="@miller")) + post_id = pg_loom.insert_one( + Post, + values=[ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ], + ) assert userId == 1 assert post_id == 1 conn.close() @@ -64,9 +68,9 @@ def test_insetting_multiple_document(self): UpdatedAtColumn, TableColumn, ForeignKeyColumn, + ColumnValue, ) from dataloom.keys import PgConfig - from typing import Optional pg_loom = Dataloom( dialect="postgres", @@ -76,7 +80,7 @@ def test_insetting_multiple_document(self): ) class User(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="users") + __tablename__: 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) @@ -86,7 +90,7 @@ class User(Model): updatedAt = UpdatedAtColumn() class Post(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="posts") + __tablename__: TableColumn = TableColumn(name="posts") id = PrimaryKeyColumn( type="int", auto_increment=True, nullable=False, unique=True ) @@ -101,10 +105,15 @@ class Post(Model): ) conn, _ = pg_loom.connect_and_sync([Post, User], drop=True, force=True) - user = User(username="@miller") - userId = pg_loom.insert_one(user) - post = Post(title="What are you doing?", userId=userId) - count = pg_loom.insert_bulk([post for i in range(5)]) + userId = pg_loom.insert_one( + User, values=ColumnValue(name="username", value="@miller") + ) + values = [ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ] + + count = pg_loom.insert_bulk(Post, values=[values for i in range(5)]) assert count == 5 conn.close() @@ -117,6 +126,8 @@ def test_relational_instances(self): PrimaryKeyColumn, Column, Dataloom, + ColumnValue, + TableColumn, ) from dataloom.keys import PgConfig @@ -128,7 +139,7 @@ def test_relational_instances(self): ) class User(Model): - __tablename__ = "users" + __tablename__: TableColumn = TableColumn(name="users") id = PrimaryKeyColumn(type="bigint", auto_increment=True) username = Column(type="text", nullable=False) name = Column(type="varchar", unique=False, length=255) @@ -136,7 +147,7 @@ class User(Model): updatedAt = UpdatedAtColumn() class Post(Model): - __tablename__ = "posts" + __tablename__: TableColumn = TableColumn(name="posts") id = PrimaryKeyColumn(type="bigint", auto_increment=True) title = Column(type="text", nullable=False, default="Hello there!!") @@ -147,11 +158,113 @@ class Post(Model): ) conn, _ = pg_loom.connect_and_sync([User, Post], drop=True, force=True) - user = User(name="Crispen", username="heyy") - userId = pg_loom.insert_one(user) - postId = pg_loom.insert_one( - Post(userId=userId, title="What are you thinking"), - ) + + userId = pg_loom.insert_one(User, ColumnValue(name="username", value="@miller")) + values = [ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ] + + postId = pg_loom.insert_one(Post, values) now = pg_loom.find_by_pk(Post, postId) assert userId == now["userId"] conn.close() + + def test_insert_bulk_with_errors(self): + from dataloom import ( + Dataloom, + Model, + Column, + PrimaryKeyColumn, + CreatedAtColumn, + UpdatedAtColumn, + TableColumn, + ForeignKeyColumn, + ColumnValue, + InvalidColumnValuesException, + ) + from dataloom.keys import PgConfig + import pytest + + pg_loom = Dataloom( + dialect="postgres", + database=PgConfig.database, + password=PgConfig.password, + user=PgConfig.user, + ) + + class User(Model): + __tablename__: 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__: 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, _ = pg_loom.connect_and_sync([Post, User], drop=True, force=True) + userId = pg_loom.insert_one( + User, values=ColumnValue(name="username", value="@miller") + ) + values = [ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ] + with pytest.raises(InvalidColumnValuesException) as exc_info: + pg_loom.insert_bulk(Post, values=values) + + assert ( + str(exc_info.value) + == "The insert_bulk method takes in values as lists of lists." + ) + + with pytest.raises(InvalidColumnValuesException) as exc_info: + pg_loom.insert_bulk( + Post, + values=[ + [ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ], + ColumnValue(name="title", value="What are you doing?"), + ], + ) + assert ( + str(exc_info.value) + == "The insert_bulk method takes in values as lists of lists." + ) + + with pytest.raises(InvalidColumnValuesException) as exc_info: + pg_loom.insert_bulk( + Post, + values=[ + [ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ], + [ + ColumnValue(name="title", value="What are you doing?"), + ], + ], + ) + assert ( + str(exc_info.value) + == "The insert_bulk method takes in values as lists of lists with equal ColumnValues." + ) + conn.close() diff --git a/dataloom/tests/postgres/test_query_pg.py b/dataloom/tests/postgres/test_query_pg.py index 82c9b51..642ea68 100644 --- a/dataloom/tests/postgres/test_query_pg.py +++ b/dataloom/tests/postgres/test_query_pg.py @@ -10,6 +10,7 @@ def test_find_by_pk_fn(self): TableColumn, ForeignKeyColumn, UnknownColumnException, + ColumnValue, ) import pytest from dataloom.keys import PgConfig @@ -44,10 +45,12 @@ class Post(Model): ) conn, _ = pg_loom.connect_and_sync([Post, User], drop=True, force=True) - user = User(username="@miller") - userId = pg_loom.insert_one(user) - post = Post(title="What are you doing?", userId=userId) - _ = pg_loom.insert_bulk([post for i in range(5)]) + userId = pg_loom.insert_one(User, ColumnValue(name="username", value="@miller")) + values = [ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ] + _ = pg_loom.insert_bulk(Post, [values for i in range(5)]) me = pg_loom.find_by_pk(User, 1) her = pg_loom.find_by_pk(User, 2) @@ -76,6 +79,7 @@ def test_find_all_fn(self): TableColumn, ForeignKeyColumn, UnknownColumnException, + ColumnValue, ) from dataloom.keys import PgConfig from typing import Optional @@ -110,10 +114,12 @@ class Post(Model): ) conn, _ = pg_loom.connect_and_sync([Post, User], drop=True, force=True) - user = User(username="@miller") - userId = pg_loom.insert_one(user) - post = Post(title="What are you doing?", userId=userId) - _ = pg_loom.insert_bulk([post for i in range(5)]) + userId = pg_loom.insert_one(User, ColumnValue(name="username", value="@miller")) + values = [ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ] + _ = pg_loom.insert_bulk(Post, [values for i in range(5)]) users = pg_loom.find_all(User) posts = pg_loom.find_all(Post) paginated = pg_loom.find_all( @@ -149,6 +155,7 @@ def test_find_one_fn(self): ForeignKeyColumn, UnknownColumnException, Filter, + ColumnValue, ) from dataloom.keys import PgConfig from typing import Optional @@ -183,10 +190,12 @@ class Post(Model): ) conn, _ = pg_loom.connect_and_sync([Post, User], drop=True, force=True) - user = User(username="@miller") - userId = pg_loom.insert_one(user) - post = Post(title="What are you doing?", userId=userId) - _ = pg_loom.insert_bulk([post for i in range(5)]) + userId = pg_loom.insert_one(User, ColumnValue(name="username", value="@miller")) + values = [ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ] + _ = pg_loom.insert_bulk(Post, [values for i in range(5)]) one_0 = pg_loom.find_one(User, filters=Filter(column="id", value=5)) one_1 = pg_loom.find_one(User, filters=Filter(column="id", value=1)) @@ -249,6 +258,7 @@ def test_find_many(self): ForeignKeyColumn, UnknownColumnException, Filter, + ColumnValue, ) from dataloom.keys import PgConfig from typing import Optional @@ -283,10 +293,12 @@ class Post(Model): ) conn, _ = pg_loom.connect_and_sync([Post, User], drop=True, force=True) - user = User(username="@miller") - userId = pg_loom.insert_one(user) - post = Post(title="What are you doing?", userId=userId) - rows = pg_loom.insert_bulk([post for i in range(5)]) + userId = pg_loom.insert_one(User, ColumnValue(name="username", value="@miller")) + values = [ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ] + rows = pg_loom.insert_bulk(Post, [values for i in range(5)]) posts = pg_loom.find_many( Post, filters=[Filter(column="id", value=1), Filter(column="userId", value=1)], diff --git a/dataloom/tests/postgres/test_update_pg.py b/dataloom/tests/postgres/test_update_pg.py index 0c2fbd0..e96d08c 100644 --- a/dataloom/tests/postgres/test_update_pg.py +++ b/dataloom/tests/postgres/test_update_pg.py @@ -1,7 +1,6 @@ class TestUpdateOnPG: def test_update_by_pk_fn(self): import time - from typing import Optional from dataloom import ( Column, @@ -24,7 +23,7 @@ def test_update_by_pk_fn(self): ) class User(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="users") + __tablename__: 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) @@ -34,7 +33,7 @@ class User(Model): updatedAt = UpdatedAtColumn() class Post(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="posts") + __tablename__: TableColumn = TableColumn(name="posts") id = PrimaryKeyColumn( type="int", auto_increment=True, nullable=False, unique=True ) @@ -50,10 +49,14 @@ class Post(Model): conn, _ = pg_loom.connect_and_sync([Post, User], drop=True, force=True) - user = User(username="@miller") - userId = pg_loom.insert_one(user) - post = Post(title="What are you doing?", userId=userId) - _ = pg_loom.insert_bulk([post for i in range(5)]) + userId = pg_loom.insert_one( + User, values=ColumnValue(name="username", value="@miller") + ) + post = [ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ] + _ = pg_loom.insert_bulk(Post, [post for i in range(5)]) time.sleep(0.05) res_1 = pg_loom.update_by_pk( User, userId, ColumnValue(name="username", value="Gari") @@ -70,7 +73,6 @@ class Post(Model): def test_update_one_fn(self): import time - from typing import Optional import pytest @@ -97,7 +99,7 @@ def test_update_one_fn(self): ) class User(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="users") + __tablename__: 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) @@ -107,7 +109,7 @@ class User(Model): updatedAt = UpdatedAtColumn() class Post(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="posts") + __tablename__: TableColumn = TableColumn(name="posts") id = PrimaryKeyColumn( type="int", auto_increment=True, nullable=False, unique=True ) @@ -122,10 +124,14 @@ class Post(Model): conn, _ = pg_loom.connect_and_sync([Post, User], drop=True, force=True) - user = User(username="@miller") - userId = pg_loom.insert_one(user) - post = Post(title="What are you doing?", userId=userId) - _ = pg_loom.insert_bulk([post for i in range(5)]) + userId = pg_loom.insert_one( + User, values=ColumnValue(name="username", value="@miller") + ) + post = [ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ] + _ = pg_loom.insert_bulk(Post, [post for i in range(5)]) time.sleep(0.05) res_1 = pg_loom.update_one( User, @@ -160,8 +166,6 @@ class Post(Model): conn.close() def test_update_bulk_fn(self): - from typing import Optional - import pytest from dataloom import ( @@ -187,7 +191,7 @@ def test_update_bulk_fn(self): ) class User(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="users") + __tablename__: 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) @@ -197,7 +201,7 @@ class User(Model): updatedAt = UpdatedAtColumn() class Post(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="posts") + __tablename__: TableColumn = TableColumn(name="posts") id = PrimaryKeyColumn( type="int", auto_increment=True, nullable=False, unique=True ) @@ -212,10 +216,14 @@ class Post(Model): conn, _ = pg_loom.connect_and_sync([Post, User], drop=True, force=True) - user = User(username="@miller") - userId = pg_loom.insert_one(user) - post = Post(title="What are you doing?", userId=userId) - _ = pg_loom.insert_bulk([post for i in range(5)]) + userId = pg_loom.insert_one( + User, values=ColumnValue(name="username", value="@miller") + ) + post = [ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ] + _ = pg_loom.insert_bulk(Post, [post for i in range(5)]) res_1 = pg_loom.update_bulk( Post, Filter(column="userId", value=userId), @@ -253,7 +261,6 @@ class Post(Model): conn.close() def test_increment_fn(self): - from typing import Optional import pytest from dataloom import ( Column, @@ -276,7 +283,7 @@ def test_increment_fn(self): ) class User(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="users") + __tablename__: 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) @@ -287,8 +294,9 @@ class User(Model): conn, _ = pg_loom.connect_and_sync([User], drop=True, force=True) - user = User(username="@miller") - userId = pg_loom.insert_one(user) + userId = pg_loom.insert_one( + User, values=ColumnValue(name="username", value="@miller") + ) affected_rows = pg_loom.increment( User, @@ -320,7 +328,6 @@ class User(Model): conn.close() def test_decrement_fn(self): - from typing import Optional import pytest from dataloom import ( Column, @@ -343,7 +350,7 @@ def test_decrement_fn(self): ) class User(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="users") + __tablename__: 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) @@ -354,8 +361,9 @@ class User(Model): conn, _ = pg_loom.connect_and_sync([User], drop=True, force=True) - user = User(username="@miller") - userId = pg_loom.insert_one(user) + userId = pg_loom.insert_one( + User, values=ColumnValue(name="username", value="@miller") + ) affected_rows = pg_loom.decrement( User, diff --git a/dataloom/tests/sqlite3/test_delete_sqlite.py b/dataloom/tests/sqlite3/test_delete_sqlite.py index 61a58d8..c8b2db1 100644 --- a/dataloom/tests/sqlite3/test_delete_sqlite.py +++ b/dataloom/tests/sqlite3/test_delete_sqlite.py @@ -1,4 +1,4 @@ -class TestDeletingOnSqlite: +class TestDeletingOnSQLite: def test_delete_by_pk_fn(self): from dataloom import ( Column, @@ -9,6 +9,7 @@ def test_delete_by_pk_fn(self): CreatedAtColumn, UpdatedAtColumn, ForeignKeyColumn, + ColumnValue, ) from typing import Optional @@ -42,8 +43,10 @@ class Post(Model): conn, _ = sqlite_loom.connect_and_sync([Post, User], drop=True, force=True) - user = User(name="Crispen", username="heyy") - userId = sqlite_loom.insert_one(user) + userId = sqlite_loom.insert_one( + User, ColumnValue(name="username", value="@miller") + ) + affected_rows_1 = sqlite_loom.delete_by_pk(User, userId) affected_rows_2 = sqlite_loom.delete_by_pk(User, 89) assert affected_rows_1 == 1 @@ -63,6 +66,7 @@ def test_delete_one_fn(self): ForeignKeyColumn, UnknownColumnException, Filter, + ColumnValue, ) from typing import Optional @@ -96,11 +100,21 @@ class Post(Model): conn, _ = sqlite_loom.connect_and_sync([Post, User], drop=True, force=True) sqlite_loom.insert_bulk( - [ - User(name="Crispen", username="heyy"), - User(name="Crispen", username="who"), - User(name="Crispen", username="hi"), - ] + User, + values=[ + [ + ColumnValue(name="name", value="Crispen"), + ColumnValue(name="username", value="hi"), + ], + [ + ColumnValue(name="name", value="Crispen"), + ColumnValue(name="username", value="heyy"), + ], + [ + ColumnValue(name="name", value="Crispen"), + ColumnValue(name="username", value="who"), + ], + ], ) sqlite_loom.delete_one(User, filters=[Filter(column="name", value="Crispen")]) rows_1 = sqlite_loom.find_many( @@ -155,6 +169,7 @@ def test_delete_bulk_fn(self): ForeignKeyColumn, UnknownColumnException, Filter, + ColumnValue, ) from typing import Optional @@ -188,11 +203,21 @@ class Post(Model): conn, _ = sqlite_loom.connect_and_sync([Post, User], drop=True, force=True) sqlite_loom.insert_bulk( - [ - User(name="Crispen", username="hi"), - User(name="Crispen", username="heyy"), - User(name="Crispen", username="hie"), - ] + User, + values=[ + [ + ColumnValue(name="name", value="Crispen"), + ColumnValue(name="username", value="hi"), + ], + [ + ColumnValue(name="name", value="Crispen"), + ColumnValue(name="username", value="heyy"), + ], + [ + ColumnValue(name="name", value="Crispen"), + ColumnValue(name="username", value="hie"), + ], + ], ) sqlite_loom.delete_bulk( User, filters=Filter(column="name", value="Crispen", operator="eq") @@ -201,11 +226,21 @@ class Post(Model): User, filters=Filter(column="name", value="Crispen", operator="eq") ) sqlite_loom.insert_bulk( - [ - User(name="Crispen", username="hi"), - User(name="Crispen", username="heyy"), - User(name="Crispen", username="hie"), - ] + User, + values=[ + [ + ColumnValue(name="name", value="Crispen"), + ColumnValue(name="username", value="hi"), + ], + [ + ColumnValue(name="name", value="Crispen"), + ColumnValue(name="username", value="heyy"), + ], + [ + ColumnValue(name="name", value="Crispen"), + ColumnValue(name="username", value="hie"), + ], + ], ) sqlite_loom.delete_bulk( User, diff --git a/dataloom/tests/sqlite3/test_insert_sqlite.py b/dataloom/tests/sqlite3/test_insert_sqlite.py index 41a8ba6..e980fcb 100644 --- a/dataloom/tests/sqlite3/test_insert_sqlite.py +++ b/dataloom/tests/sqlite3/test_insert_sqlite.py @@ -1,4 +1,4 @@ -class TestInsertingOnSQLite: +class TestInsertingOnPG: def test_insetting_single_document(self): from dataloom import ( Dataloom, @@ -9,14 +9,13 @@ def test_insetting_single_document(self): UpdatedAtColumn, TableColumn, ForeignKeyColumn, + ColumnValue, ) - from typing import Optional - sqlite_loom = Dataloom(dialect="sqlite", database="hi.db") class User(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="users") + __tablename__: 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) @@ -26,7 +25,7 @@ class User(Model): updatedAt = UpdatedAtColumn() class Post(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="posts") + __tablename__: TableColumn = TableColumn(name="posts") id = PrimaryKeyColumn( type="int", auto_increment=True, nullable=False, unique=True ) @@ -41,10 +40,16 @@ class Post(Model): ) conn, _ = sqlite_loom.connect_and_sync([Post, User], drop=True, force=True) - user = User(username="@miller") - userId = sqlite_loom.insert_one(user) - post = Post(title="What are you doing?", userId=userId) - post_id = sqlite_loom.insert_one(post) + userId = sqlite_loom.insert_one( + User, ColumnValue(name="username", value="@miller") + ) + post_id = sqlite_loom.insert_one( + Post, + values=[ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ], + ) assert userId == 1 assert post_id == 1 conn.close() @@ -59,13 +64,13 @@ def test_insetting_multiple_document(self): UpdatedAtColumn, TableColumn, ForeignKeyColumn, + ColumnValue, ) - from typing import Optional sqlite_loom = Dataloom(dialect="sqlite", database="hi.db") class User(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="users") + __tablename__: 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) @@ -75,7 +80,7 @@ class User(Model): updatedAt = UpdatedAtColumn() class Post(Model): - __tablename__: Optional[TableColumn] = TableColumn(name="posts") + __tablename__: TableColumn = TableColumn(name="posts") id = PrimaryKeyColumn( type="int", auto_increment=True, nullable=False, unique=True ) @@ -90,10 +95,15 @@ class Post(Model): ) conn, _ = sqlite_loom.connect_and_sync([Post, User], drop=True, force=True) - user = User(username="@miller") - userId = sqlite_loom.insert_one(user) - post = Post(title="What are you doing?", userId=userId) - count = sqlite_loom.insert_bulk([post for i in range(5)]) + userId = sqlite_loom.insert_one( + User, values=ColumnValue(name="username", value="@miller") + ) + values = [ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ] + + count = sqlite_loom.insert_bulk(Post, values=[values for i in range(5)]) assert count == 5 conn.close() @@ -106,12 +116,14 @@ def test_relational_instances(self): PrimaryKeyColumn, Column, Dataloom, + ColumnValue, + TableColumn, ) sqlite_loom = Dataloom(dialect="sqlite", database="hi.db") class User(Model): - __tablename__ = "users" + __tablename__: TableColumn = TableColumn(name="users") id = PrimaryKeyColumn(type="int", auto_increment=True) username = Column(type="text", nullable=False) name = Column(type="varchar", unique=False, length=255) @@ -119,7 +131,7 @@ class User(Model): updatedAt = UpdatedAtColumn() class Post(Model): - __tablename__ = "posts" + __tablename__: TableColumn = TableColumn(name="posts") id = PrimaryKeyColumn(type="int", auto_increment=True) title = Column(type="text", nullable=False, default="Hello there!!") @@ -130,11 +142,109 @@ class Post(Model): ) conn, _ = sqlite_loom.connect_and_sync([User, Post], drop=True, force=True) - user = User(name="Crispen", username="heyy") - userId = sqlite_loom.insert_one(user) - postId = sqlite_loom.insert_one( - Post(userId=userId, title="What are you thinking"), + + userId = sqlite_loom.insert_one( + User, ColumnValue(name="username", value="@miller") ) + values = [ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ] + + postId = sqlite_loom.insert_one(Post, values) now = sqlite_loom.find_by_pk(Post, postId) assert userId == now["userId"] + + def test_insert_bulk_with_errors(self): + from dataloom import ( + Dataloom, + Model, + Column, + PrimaryKeyColumn, + CreatedAtColumn, + UpdatedAtColumn, + TableColumn, + ForeignKeyColumn, + ColumnValue, + InvalidColumnValuesException, + ) + + import pytest + + sqlite_loom = Dataloom(dialect="sqlite", database="hi.db") + + class User(Model): + __tablename__: 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__: 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, _ = sqlite_loom.connect_and_sync([Post, User], drop=True, force=True) + userId = sqlite_loom.insert_one( + User, values=ColumnValue(name="username", value="@miller") + ) + values = [ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ] + with pytest.raises(InvalidColumnValuesException) as exc_info: + sqlite_loom.insert_bulk(Post, values=values) + + assert ( + str(exc_info.value) + == "The insert_bulk method takes in values as lists of lists." + ) + + with pytest.raises(InvalidColumnValuesException) as exc_info: + sqlite_loom.insert_bulk( + Post, + values=[ + [ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ], + ColumnValue(name="title", value="What are you doing?"), + ], + ) + assert ( + str(exc_info.value) + == "The insert_bulk method takes in values as lists of lists." + ) + + with pytest.raises(InvalidColumnValuesException) as exc_info: + sqlite_loom.insert_bulk( + Post, + values=[ + [ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ], + [ + ColumnValue(name="title", value="What are you doing?"), + ], + ], + ) + assert ( + str(exc_info.value) + == "The insert_bulk method takes in values as lists of lists with equal ColumnValues." + ) conn.close() diff --git a/dataloom/tests/sqlite3/test_query_sqlite.py b/dataloom/tests/sqlite3/test_query_sqlite.py index c814630..e481a06 100644 --- a/dataloom/tests/sqlite3/test_query_sqlite.py +++ b/dataloom/tests/sqlite3/test_query_sqlite.py @@ -10,6 +10,7 @@ def test_find_by_pk_fn(self): TableColumn, ForeignKeyColumn, UnknownColumnException, + ColumnValue, ) import pytest from typing import Optional @@ -38,10 +39,14 @@ class Post(Model): ) conn, _ = sqlite_loom.connect_and_sync([Post, User], drop=True, force=True) - user = User(username="@miller") - userId = sqlite_loom.insert_one(user) - post = Post(title="What are you doing?", userId=userId) - _ = sqlite_loom.insert_bulk([post for i in range(5)]) + userId = sqlite_loom.insert_one( + User, ColumnValue(name="username", value="@miller") + ) + values = [ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ] + _ = sqlite_loom.insert_bulk(Post, [values for i in range(5)]) me = sqlite_loom.find_by_pk(User, 1) her = sqlite_loom.find_by_pk(User, 2) @@ -70,6 +75,7 @@ def test_find_all_fn(self): TableColumn, ForeignKeyColumn, UnknownColumnException, + ColumnValue, ) from typing import Optional import pytest @@ -98,10 +104,14 @@ class Post(Model): ) conn, _ = sqlite_loom.connect_and_sync([Post, User], drop=True, force=True) - user = User(username="@miller") - userId = sqlite_loom.insert_one(user) - post = Post(title="What are you doing?", userId=userId) - _ = sqlite_loom.insert_bulk([post for i in range(5)]) + userId = sqlite_loom.insert_one( + User, ColumnValue(name="username", value="@miller") + ) + values = [ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ] + _ = sqlite_loom.insert_bulk(Post, [values for i in range(5)]) users = sqlite_loom.find_all(User) posts = sqlite_loom.find_all(Post) paginated = sqlite_loom.find_all( @@ -137,6 +147,7 @@ def test_find_one_fn(self): ForeignKeyColumn, UnknownColumnException, Filter, + ColumnValue, ) from typing import Optional import pytest @@ -165,10 +176,14 @@ class Post(Model): ) conn, _ = sqlite_loom.connect_and_sync([Post, User], drop=True, force=True) - user = User(username="@miller") - userId = sqlite_loom.insert_one(user) - post = Post(title="What are you doing?", userId=userId) - _ = sqlite_loom.insert_bulk([post for i in range(5)]) + userId = sqlite_loom.insert_one( + User, ColumnValue(name="username", value="@miller") + ) + values = [ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ] + _ = sqlite_loom.insert_bulk(Post, [values for i in range(5)]) one_0 = sqlite_loom.find_one(User, filters=Filter(column="id", value=5)) one_1 = sqlite_loom.find_one(User, filters=Filter(column="id", value=1)) @@ -231,6 +246,7 @@ def test_find_many(self): ForeignKeyColumn, UnknownColumnException, Filter, + ColumnValue, ) from typing import Optional import pytest @@ -259,10 +275,14 @@ class Post(Model): ) conn, _ = sqlite_loom.connect_and_sync([Post, User], drop=True, force=True) - user = User(username="@miller") - userId = sqlite_loom.insert_one(user) - post = Post(title="What are you doing?", userId=userId) - rows = sqlite_loom.insert_bulk([post for i in range(5)]) + userId = sqlite_loom.insert_one( + User, ColumnValue(name="username", value="@miller") + ) + values = [ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ] + rows = sqlite_loom.insert_bulk(Post, [values for i in range(5)]) posts = sqlite_loom.find_many( Post, filters=[Filter(column="id", value=1), Filter(column="userId", value=1)], diff --git a/dataloom/tests/sqlite3/test_update_sqlite.py b/dataloom/tests/sqlite3/test_update_sqlite.py index bebecaf..4c93bb7 100644 --- a/dataloom/tests/sqlite3/test_update_sqlite.py +++ b/dataloom/tests/sqlite3/test_update_sqlite.py @@ -43,11 +43,14 @@ class Post(Model): ) conn, _ = sqlite_loom.connect_and_sync([Post, User], drop=True, force=True) - - user = User(username="@miller") - userId = sqlite_loom.insert_one(user) - post = Post(title="What are you doing?", userId=userId) - _ = sqlite_loom.insert_bulk([post for i in range(5)]) + userId = sqlite_loom.insert_one( + User, values=ColumnValue(name="username", value="@miller") + ) + post = [ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ] + _ = sqlite_loom.insert_bulk(Post, values=[post for i in range(5)]) time.sleep(0.05) res_1 = sqlite_loom.update_by_pk( User, userId, ColumnValue(name="username", value="Gari") @@ -110,10 +113,14 @@ class Post(Model): conn, _ = sqlite_loom.connect_and_sync([Post, User], drop=True, force=True) - user = User(username="@miller") - userId = sqlite_loom.insert_one(user) - post = Post(title="What are you doing?", userId=userId) - _ = sqlite_loom.insert_bulk([post for i in range(5)]) + userId = sqlite_loom.insert_one( + User, values=ColumnValue(name="username", value="@miller") + ) + post = [ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ] + _ = sqlite_loom.insert_bulk(Post, values=[post for i in range(5)]) time.sleep(0.05) res_1 = sqlite_loom.update_one( User, @@ -192,10 +199,14 @@ class Post(Model): conn, _ = sqlite_loom.connect_and_sync([Post, User], drop=True, force=True) - user = User(username="@miller") - userId = sqlite_loom.insert_one(user) - post = Post(title="What are you doing?", userId=userId) - _ = sqlite_loom.insert_bulk([post for i in range(5)]) + userId = sqlite_loom.insert_one( + User, values=ColumnValue(name="username", value="@miller") + ) + post = [ + ColumnValue(name="title", value="What are you doing?"), + ColumnValue(name="userId", value=userId), + ] + _ = sqlite_loom.insert_bulk(Post, values=[post for i in range(5)]) res_1 = sqlite_loom.update_bulk( Post, Filter(column="userId", value=userId), @@ -254,8 +265,9 @@ class User(Model): conn, _ = sqlite_loom.connect_and_sync([User], drop=True, force=True) - user = User(username="@miller") - userId = sqlite_loom.insert_one(user) + userId = sqlite_loom.insert_one( + User, values=ColumnValue(name="username", value="@miller") + ) affected_rows = sqlite_loom.increment( User, @@ -315,8 +327,9 @@ class User(Model): conn, _ = sqlite_loom.connect_and_sync([User], drop=True, force=True) - user = User(username="@miller") - userId = sqlite_loom.insert_one(user) + userId = sqlite_loom.insert_one( + User, values=ColumnValue(name="username", value="@miller") + ) affected_rows = sqlite_loom.decrement( User, diff --git a/dataloom/utils/__init__.py b/dataloom/utils/__init__.py index ccf4066..38b457e 100644 --- a/dataloom/utils/__init__.py +++ b/dataloom/utils/__init__.py @@ -410,6 +410,21 @@ def get_formatted_query( return sql +def get_null_field_placeholder( + column: str, + dialect: DIALECT_LITERAL, + table_name: str, + pk_name: str, +) -> str: + if dialect == "mysql": + value = f"IFNULL(%s, DEFAULT(`{column}`))" + if dialect == "sqlite": + value = "?" + if dialect == "postgres": + value = f"""COALESCE(%s, DEFAULT("{column}"))""" + return value + + def get_insert_bulk_attrs( instance, dialect: DIALECT_LITERAL, @@ -418,16 +433,18 @@ def get_insert_bulk_attrs( fields, pk_name, fks, updatedAtColumName = get_table_fields( instance, dialect=dialect ) + placeholders, column_values, column_names = get_column_values( table_name=instance._get_table_name(), dialect=dialect, fields=fields, values=values, ) + _placeholders = ", ".join( - ["?" if dialect == "sqlite" else "%s" for _ in placeholders] + ["?" if dialect == "sqlite" else "%s" for f in column_names] ) _column_names = ", ".join( [f'"{f}"' if dialect == "postgres" else f"`{f}`" for f in column_names] ) - return _column_names, _placeholders, column_values + return (_column_names, _placeholders, column_values) diff --git a/hi.db b/hi.db index f7f31869acb457265e5a39b1f14f6e9917f779f2..c30cb97ad3524133f33af511801be4c7ceddc9b5 100644 GIT binary patch delta 772 zcma)(Id2m|6vt=p!@HgZX&K)~2!xGraMpk!5U^;>QIL)zpomD80WCPfPUj-KM9=|Y zBpQ(74*&((4s2+$B1jbE8w62+WE!Lh8U)PlCNd2XBhAdy@BiNa&9Ts77COws#O_0N z0Dwy7*2vE+9W?h7I(FlODq27h+H|xU`4JI>w!AFmWJ`^UYteNPi<@p<`^33*UZOO+ z`s%uX5Tcx$E!*nF@N4PosQ3nq2uRnd#b)d3tW}$9G+T50YMT#w@m?Yf3@?p0l&kTc!2 zd_i#>!%GDoi{r%m4t^>uyT4!sM$5TkdBQwCapWi|If*mbIId;&`T(a{UIcncI^Vc< z%Pg9Q3q`X~{7Ytbmgya3{4uM`1GAMJr9pBI8ukI~!|!ktzIC!S!*7DX{2bN5!i=I9jN$^ c1vGyFVhE=={PThMj~w|wa+K{HmSH6652^dQ;s5{u delta 829 zcmaKnPiPZC6vk(=o88RrCNGtSZS9|IYH1s>vPslRRcI|z|DerHq+p5ZHdzfxOj58Y z*rW%+gSO65#3Bg6p6uRSD&9mC6wzB1MWG^s6pEKBI*Enk5IVf!!S_AB-+Qx*-0UJZ zzjH_L8HA9V{jJw_Z(&ckhpBrnzHyVsND(~_&!`-OUF3=M$aRR^aJS0)MUPCR4_3hY z#c3r3g6VbQqFF_t&*V-z0r&aRar>y%O-^8kJwh{5pYjfoDxsLNhv=;MgX^`2nbA(4 zN{YO`Wd`jj`qI_zoz5G%{6x6g#Gq}|iJ__}1noEzgM3jb$LXegs}J|7YH;bipdL{% zGSidBmHg$DdDZR`RaGrH?a4~Vb0%H&%+Zf>3%v_*zO>uEV0E~!VFk{io0YlJcyWq; z68v1StW~J~|Ep|K*#%##apmz^TtJMJCk1shKwrCBXpgM&Z0sR@q4?=v0q7}r&iVtx z;5nd$^uf@^sL36?;6lC#IhjeO!+LmMzaH+_>uhd&+p(=pKl^=B zEo@oCUVvZ72{wi`ScSJ>TVDMYbc@B)oR3t_cDfy0fW%`3Baz9OMxL!L`A9aCFDwtO zK`ks~t&eqit>)-^px-j>0)`C)8&*^FJ17$7P+P)C8@X8OLDYL^p64+fW2Nn+bz_>EXeR#WyT(9PR~p;r*BSOLv~v0)ZH5Q&hp i$yh;)ecCYq diff --git a/playground.py b/playground.py index 03801cf..309ac1f 100644 --- a/playground.py +++ b/playground.py @@ -30,11 +30,13 @@ host="localhost", logs_filename="logs.sql", port=3306, + sql_logger="console", ) sqlite_loom = Dataloom( dialect="sqlite", database="hi.db", logs_filename="sqlite-logs.sql", + sql_logger="console", ) @@ -78,12 +80,18 @@ class Post(Model): ColumnValue(name="username", value="@miller"), ColumnValue(name="name", value="Jonh"), ], - ColumnValue(name="username", value="@brown"), - ColumnValue(name="username", value="@blue"), + [ + ColumnValue(name="username", value="@brown"), + ColumnValue(name="name", value="Jonh"), + ], + [ + ColumnValue(name="username", value="@blue"), + ColumnValue(name="name", value="Jonh"), + ], ], ) -# pg_loom.decrement( +# sqlite_loom.decrement( # User, # filters=Filter(column="id", value=1), # column=ColumnValue(name="tokenVersion", value=2.6), @@ -91,11 +99,11 @@ class Post(Model): # cate = Category(name="general") -# categoryId = pg_loom.insert_one(cate) +# categoryId = mysql_loom.insert_one(cate) # post = Post(title="What are you doing?", userId=userId, categoryId=categoryId) -# post_id = pg_loom.insert_bulk([post for i in range(5)]) +# post_id = mysql_loom.insert_bulk([post for i in range(5)]) -# post = pg_loom.find_one( +# post = mysql_loom.find_one( # Post, # filters=[ # Filter(column="id", operator="eq", value=4, join_next_filter_with="AND"), @@ -115,7 +123,7 @@ class Post(Model): # ) # print(post) -# post = pg_loom.find_by_pk( +# post = mysql_loom.find_by_pk( # Post, # pk=1, # select=["id", "completed", "title", "createdAt"], @@ -130,17 +138,17 @@ class Post(Model): # return_dict=True, # ) -re = pg_loom.update_one( - Post, - values=[ - ColumnValue(name="title", value="Hey"), - ColumnValue(name="completed", value=True), - ], - filters=[ - Filter(column="id", value=1, join_next_filter_with="AND"), - Filter(column="userId", value=1, join_next_filter_with="AND"), - ], -) +# re = mysql_loom.update_one( +# Post, +# values=[ +# ColumnValue(name="title", value="Hey"), +# ColumnValue(name="completed", value=True), +# ], +# filters=[ +# Filter(column="id", value=1, join_next_filter_with="AND"), +# Filter(column="userId", value=1, join_next_filter_with="AND"), +# ], +# ) # print(post) # print(post) @@ -164,28 +172,28 @@ class Post(Model): # ) # print(posts) -posts = pg_loom.find_all( - Post, - select=["id", "completed", "title", "createdAt"], - limit=3, - offset=0, - order=[ - Order(column="createdAt", order="ASC"), - Order( - column="id", - order="DESC", - ), - ], - include=[ - Include( - model=User, - select=["id", "username", "name"], - limit=1, - offset=0, - ), - ], - return_dict=True, -) +# posts = pg_loom.find_all( +# Post, +# select=["id", "completed", "title", "createdAt"], +# limit=3, +# offset=0, +# order=[ +# Order(column="createdAt", order="ASC"), +# Order( +# column="id", +# order="DESC", +# ), +# ], +# include=[ +# Include( +# model=User, +# select=["id", "username", "name"], +# limit=1, +# offset=0, +# ), +# ], +# return_dict=True, +# ) # print(posts) # posts = pg_loom.find_many( # Post, @@ -213,7 +221,5 @@ class Post(Model): # print(posts) -print(max(9, 5, 3)) - if __name__ == "__main__": conn.close()