fixing loga tests and mysql-bug

Changelog.md

@@ -69,6 +69,7 @@ post = loom.find_all(
     column="id",
   )
   ```
+- Updated logger colors and formatting.
 
 # Dataloom **`2.1.1`**
 

dataloom/model/__init__.py

@@ -188,10 +188,8 @@ def _get_select_where_stm(
 
         if function is not None:
             select = [
-                f'{function.upper()}({"DISTINCT " if distinct else ''}{f"{column}" if dialect == 'postgres' else f"`{column}`"})'
+                f'{"DISTINCT " if distinct else ''}{f'"{column}"' if dialect == 'postgres' else f"`{column}`"}'
             ]
-            distinct = False
-
         placeholder_filters, placeholder_filter_values = get_table_filters(
             table_name=cls._get_table_name(),
             dialect=dialect,

dataloom/statements/__init__.py

@@ -229,35 +229,62 @@ def _get_select_where_command(
             "" if offset is None else f"OFFSET { offset}",
         ]
         if self.dialect == "postgres":
-            sql = PgStatements.SELECT_WHERE_COMMAND.format(
-                column_names=", ".join(fields)
-                if function is not None
-                else ", ".join([f'"{f}"' for f in fields] + group_fns),
-                table_name=f'"{self.table_name}"',
-                filters=" ".join(placeholder_filters),
-                options=" ".join(options),
-                distinct="DISTINCT" if distinct else "",
-            )
+            if function is None:
+                sql = PgStatements.SELECT_WHERE_COMMAND.format(
+                    column_names=", ".join([f'"{f}"' for f in fields] + group_fns),
+                    table_name=f'"{self.table_name}"',
+                    filters=" ".join(placeholder_filters),
+                    options=" ".join(options),
+                    distinct="DISTINCT" if distinct else "",
+                )
+            else:
+                sql = PgStatements.SELECT_WHERE_FN_COMMAND.format(
+                    fn=function.upper(),
+                    column_names=", ".join(fields),
+                    table_name=f'"{self.table_name}"',
+                    filters=" ".join(placeholder_filters),
+                    options=" ".join(options),
+                )
         elif self.dialect == "mysql":
-            sql = MySqlStatements.SELECT_WHERE_COMMAND.format(
-                column_names=", ".join(fields)
-                if function is not None
-                else ", ".join([f"`{name}`" for name in fields] + group_fns),
-                table_name=f"`{self.table_name}`",
-                filters=" ".join(placeholder_filters),
-                options=" ".join(options),
-                distinct="DISTINCT" if distinct else "",
-            )
+            if function is None:
+                sql = MySqlStatements.SELECT_WHERE_COMMAND.format(
+                    column_names=", ".join(
+                        [f"`{name}`" for name in fields] + group_fns
+                    ),
+                    table_name=f"`{self.table_name}`",
+                    filters=" ".join(placeholder_filters),
+                    options=" ".join(options),
+                    distinct="DISTINCT" if distinct else "",
+                )
+            else:
+                sql = MySqlStatements.SELECT_WHERE_FN_COMMAND.format(
+                    fn=function.upper(),
+                    column_names=", ".join(fields),
+                    table_name=f"`{self.table_name}`",
+                    filters=" ".join(placeholder_filters),
+                    options=" ".join(options),
+                )
+
         elif self.dialect == "sqlite":
-            sql = Sqlite3Statements.SELECT_WHERE_COMMAND.format(
-                column_names=", ".join(fields)
-                if function is not None
-                else ", ".join([f"`{name}`" for name in fields] + group_fns),
-                table_name=f"`{self.table_name}`",
-                filters=" ".join(placeholder_filters),
-                options=" ".join(options),
-                distinct="DISTINCT" if distinct else "",
-            )
+            if function is None:
+                sql = Sqlite3Statements.SELECT_WHERE_COMMAND.format(
+                    column_names=", ".join(
+                        [f"`{name}`" for name in fields] + group_fns
+                    ),
+                    table_name=f"`{self.table_name}`",
+                    filters=" ".join(placeholder_filters),
+                    options=" ".join(options),
+                    distinct="DISTINCT" if distinct else "",
+                )
+            else:
+                sql = Sqlite3Statements.SELECT_WHERE_FN_COMMAND.format(
+                    fn=function.upper(),
+                    column_names=", ".join(fields),
+                    table_name=f"`{self.table_name}`",
+                    filters=" ".join(placeholder_filters),
+                    options=" ".join(options),
+                )
+
         else:
             raise UnsupportedDialectException(
                 "The dialect passed is not supported the supported dialects are: {'postgres', 'mysql', 'sqlite'}"
@@ -286,32 +313,57 @@ def _get_select_command(
         ]
 
         if self.dialect == "postgres":
-            sql = PgStatements.SELECT_COMMAND.format(
-                column_names=", ".join(fields)
-                if function is not None
-                else ", ".join([f'"{name}"' for name in fields] + group_fns),
-                table_name=f'"{self.table_name}"',
-                options=" ".join(options),
-                distinct="DISTINCT" if distinct else "",
-            )
+            if function is None:
+                sql = PgStatements.SELECT_COMMAND.format(
+                    column_names=", ".join(
+                        [f'"{name}"' for name in fields] + group_fns
+                    ),
+                    table_name=f'"{self.table_name}"',
+                    options=" ".join(options),
+                    distinct="DISTINCT" if distinct else "",
+                )
+            else:
+                sql = PgStatements.SELECT_FN_COMMAND.format(
+                    fn=function.upper(),
+                    column_names=", ".join(fields),
+                    table_name=f'"{self.table_name}"',
+                    options=" ".join(options),
+                )
         elif self.dialect == "mysql":
-            sql = MySqlStatements.SELECT_COMMAND.format(
-                column_names=", ".join(fields)
-                if function is not None
-                else ", ".join([f"`{name}`" for name in fields] + group_fns),
-                table_name=f"`{self.table_name}`",
-                options=" ".join(options),
-                distinct="DISTINCT" if distinct else "",
-            )
+            if function is None:
+                sql = MySqlStatements.SELECT_COMMAND.format(
+                    column_names=", ".join(fields)
+                    if function is not None
+                    else ", ".join([f"`{name}`" for name in fields] + group_fns),
+                    table_name=f"`{self.table_name}`",
+                    options=" ".join(options),
+                    distinct="DISTINCT" if distinct else "",
+                )
+            else:
+                sql = MySqlStatements.SELECT_FN_COMMAND.format(
+                    fn=function.upper(),
+                    column_names=", ".join(fields),
+                    table_name=f"`{self.table_name}`",
+                    options=" ".join(options),
+                )
+
         elif self.dialect == "sqlite":
-            sql = Sqlite3Statements.SELECT_COMMAND.format(
-                column_names=", ".join(fields)
-                if function is not None
-                else ", ".join([f"`{name}`" for name in fields] + group_fns),
-                table_name=f"`{self.table_name}`",
-                options=" ".join(options),
-                distinct="DISTINCT" if distinct else "",
-            )
+            if function is None:
+                sql = Sqlite3Statements.SELECT_COMMAND.format(
+                    column_names=", ".join(
+                        [f"`{name}`" for name in fields] + group_fns
+                    ),
+                    table_name=f"`{self.table_name}`",
+                    options=" ".join(options),
+                    distinct="DISTINCT" if distinct else "",
+                )
+            else:
+                sql = Sqlite3Statements.SELECT_FN_COMMAND.format(
+                    fn=function.upper(),
+                    column_names=", ".join(fields),
+                    table_name=f"`{self.table_name}`",
+                    options=" ".join(options),
+                )
         else:
             raise UnsupportedDialectException(
                 "The dialect passed is not supported the supported dialects are: {'postgres', 'mysql', 'sqlite'}"

dataloom/statements/statements.py

@@ -92,6 +92,21 @@ class MySqlStatements:
     )
     SELECT_BY_PK = "SELECT {column_names} FROM {table_name} WHERE {pk_name} = {pk};"
     SELECT_WHERE_COMMAND = "SELECT {distinct} {column_names} FROM {table_name} WHERE {filters} {options};".strip()
+    SELECT_WHERE_FN_COMMAND = """
+    SELECT {fn}({column_names}) FROM (
+            SELECT {column_names}
+            FROM {table_name}
+            WHERE {filters}
+            {options}
+    ) AS subquery;
+    """
+    SELECT_FN_COMMAND = """
+    SELECT {fn}({column_names}) FROM (
+            SELECT {column_names}
+            FROM {table_name}
+            {options}
+    ) AS subquery;
+    """
 
     # ------------- child parent bidirectional sub queries
     SELECT_CHILD_BY_PK = """
@@ -219,6 +234,21 @@ class Sqlite3Statements:
     )
     SELECT_BY_PK = "SELECT {column_names} FROM {table_name} WHERE {pk_name} = {pk};"
     SELECT_WHERE_COMMAND = "SELECT {distinct} {column_names} FROM {table_name} WHERE {filters} {options};".strip()
+    SELECT_WHERE_FN_COMMAND = """
+    SELECT {fn}({column_names}) FROM (
+            SELECT {column_names}
+            FROM {table_name}
+            WHERE {filters}
+            {options}
+    ) AS subquery;
+    """
+    SELECT_FN_COMMAND = """
+    SELECT {fn}({column_names}) FROM (
+            SELECT {column_names}
+            FROM {table_name}
+            {options}
+    ) AS subquery;
+    """
 
     # ------------- child parent bidirectional sub queries
     SELECT_CHILD_BY_PK = """
@@ -344,6 +374,21 @@ class PgStatements:
     )
     SELECT_BY_PK = "SELECT {column_names} FROM {table_name} WHERE {pk_name} = {pk};"
     SELECT_WHERE_COMMAND = "SELECT {distinct} {column_names} FROM {table_name} WHERE {filters} {options};".strip()
+    SELECT_WHERE_FN_COMMAND = """
+    SELECT {fn}({column_names}) FROM (
+            SELECT {column_names}
+            FROM {table_name}
+            WHERE {filters}
+            {options}
+    ) AS subquery;
+    """
+    SELECT_FN_COMMAND = """
+    SELECT {fn}({column_names}) FROM (
+            SELECT {column_names}
+            FROM {table_name}
+            {options}
+    ) AS subquery;
+    """
 
     # ------------- child parent bidirectional sub queries
     SELECT_CHILD_BY_PK = """

dataloom/tests/mysql/test_utils_fns_mysql.py

@@ -0,0 +1,99 @@
+class TestUtilsFunctionsMySQL:
+    def test_util_fns(self):
+        from dataloom import (
+            Loom,
+            Model,
+            Column,
+            PrimaryKeyColumn,
+            CreatedAtColumn,
+            UpdatedAtColumn,
+            TableColumn,
+            ForeignKeyColumn,
+            ColumnValue,
+            Filter,
+        )
+        from dataloom.keys import MySQLConfig
+        from typing import Optional
+
+        mysql_loom = Loom(
+            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)
+
+        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)
+        userId = mysql_loom.insert_one(
+            instance=User,
+            values=ColumnValue(name="username", value="@miller"),
+        )
+
+        for title in ["Hey", "Hello", "What are you doing", "Coding"]:
+            mysql_loom.insert_one(
+                instance=Post,
+                values=[
+                    ColumnValue(name="userId", value=userId),
+                    ColumnValue(name="title", value=title),
+                ],
+            )
+
+        count = mysql_loom.count(
+            instance=Post,
+            filters=Filter(
+                column="id",
+                operator="between",
+                value=[1, 7],
+            ),
+            column="id",
+            limit=3,
+            offset=0,
+            distinct=True,
+        )
+        assert count == 3
+
+        avg = mysql_loom.avg(
+            instance=Post,
+            column="id",
+            distinct=True,
+        )
+        assert avg == 2.5
+        sum = mysql_loom.sum(
+            instance=Post,
+            column="id",
+            distinct=True,
+        )
+        assert sum == 10
+        min = mysql_loom.min(
+            instance=Post,
+            column="id",
+            distinct=True,
+        )
+        assert min == 1
+        max = mysql_loom.max(
+            instance=Post,
+            column="id",
+            distinct=True,
+        )
+        assert max == 4
+        conn.close()