tests creating tables and docs

README.md

@@ -1,6 +1,18 @@
 ### dataloom
 
-An `orm` for python.
+**`dataloom`** is a lightweight and versatile Object-Relational Mapping (ORM) library for Python. With support for `PostgreSQL`, `MySQL`, and `SQLite3` databases, `dataloom` simplifies database interactions, providing a seamless experience for developers.
+
+### Key Features:
+
+- **Lightweight**: `dataloom` is designed to be minimalistic and easy to use, ensuring a streamlined `ORM` experience without unnecessary complexities.
+
+- **Database Support**: `dataloom` supports popular relational databases such as `PostgreSQL`, `MySQL`, and `SQLite3`, making it suitable for a variety of projects.
+
+- **Simplified Querying**: The `ORM` simplifies the process of database querying, allowing developers to interact with the database using Python classes and methods rather than raw SQL queries.
+
+- **Intuitive Syntax**: `dataloom`'s syntax is intuitive and `Pythonic`, making it accessible for developers familiar with the Python language.
+
+- **Flexible Data Types**: The `ORM` seamlessly handles various data types, offering flexibility in designing database schemas.
 
 ### Usage
 
@@ -11,13 +23,23 @@ In this section we are going to go through how you can use our `orm` package in
 A database connection is required. In this example a connection will be set on a `postgres` database instance that have a database name `hi.
 
 ```py
-from dataloom import Database, Model, Column
+from dataloom import (
+    Dataloom
+)
+from typing import Optional
 
-db = Database("hi", password="root", user="postgres")
-conn = db.connect()
+pg_loom = Dataloom(
+    dialect="postgres", database="hi", password="root", user="postgres", logging=True
+)
+mysql_loom = Dataloom(dialect="mysql", database="hi", password="root", user="root")
+sqlite_loom = Dataloom(dialect="sqlite", database="hi.db")
 
-if __name__ == "__main__":
+conn = pg_loom.connect()
+
+
+if __name__ == "main":
     conn.close()
+
 ```
 
 The database class takes in the following options:

dataloom/__init__.py

@@ -7,6 +7,7 @@
     ForeignKeyColumn,
     UpdatedAtColumn,
     Column,
+    TableColumn,
 )
 
 Dataloom = Dataloom

dataloom/db/__init__.py

@@ -169,6 +169,50 @@ def connect(self):
             )
         return self.conn
 
+    def connect_and_sync(
+        self, models: list[Model], drop=False, force=False, alter=False
+    ):
+        try:
+            if self.dialect == "postgres":
+                options = ConnectionOptionsFactory.get_connection_options(
+                    **self.connection_options
+                )
+                with psycopg2.connect(**options) as conn:
+                    self.conn = conn
+            elif self.dialect == "mysql":
+                options = ConnectionOptionsFactory.get_connection_options(
+                    **self.connection_options
+                )
+                self.conn = connector.connect(**options)
+
+            elif self.dialect == "sqlite":
+                options = ConnectionOptionsFactory.get_connection_options(
+                    **self.connection_options
+                )
+                if "database" in options:
+                    with sqlite3.connect(options.get("database")) as conn:
+                        self.conn = conn
+                else:
+                    with sqlite3.connect(**options) as conn:
+                        self.conn = conn
+            else:
+                raise UnsupportedDialectException(
+                    "The dialect passed is not supported the supported dialects are: {'postgres', 'mysql', 'sqlite'}"
+                )
+            for model in models:
+                if drop or force:
+                    self._execute_sql(model._drop_sql(dialect=self.dialect))
+                    self._execute_sql(model._create_sql(dialect=self.dialect))
+                elif alter:
+                    pass
+                else:
+                    self._execute_sql(
+                        model._create_sql(dialect=self.dialect, ignore_exists=True)
+                    )
+            return self.conn, self.tables
+        except Exception as e:
+            raise Exception(e)
+
     def sync(self, models: list[Model], drop=False, force=False, alter=False):
         try:
             for model in models:
@@ -267,29 +311,6 @@ def sync(self, models: list[Model], drop=False, force=False, alter=False):
 #         except Exception as e:
 #             raise Exception(e)
 
-#     def connect_and_sync(
-#         self, models: list[Model], drop=False, force=False, alter=False
-#     ):
-#         try:
-#             self.conn = psycopg2.connect(
-#                 host=self.host,
-#                 database=self.database,
-#                 user=self.user,
-#                 password=self.password,
-#                 port=self.port,
-#             )
-#             for model in models:
-#                 if drop or force:
-#                     self._execute_sql(model._drop_sql())
-#                     self._execute_sql(model._create_sql())
-#                 elif alter:
-#                     pass
-#                 else:
-#                     self._execute_sql(model._create_sql(ignore_exists=True))
-#             return self.conn, self.tables
-
-#         except Exception as e:
-#             raise Exception(e)
 
 #     def sync(self, models: list[Model], drop=False, force=False, alter=False):
 #         try:

dataloom/model/column.py

@@ -61,7 +61,7 @@ def sql_type(self, dialect: str):
 
         elif dialect == "mysql":
             if self.type in MYSQL_SQL_TYPES:
-                if self.unique and self.type == "text":
+                if (self.unique or self.default) and self.type == "text":
                     return f"{MYSQL_SQL_TYPES['varchar']}({self.length if self.length is not None else 255})"
                 return (
                     f"{MYSQL_SQL_TYPES[self.type]}({self.length})"
@@ -142,7 +142,7 @@ def sql_type(self, dialect: str):
 
         elif dialect == "mysql":
             if self.type in MYSQL_SQL_TYPES:
-                if self.unique and self.type == "text":
+                if (self.unique or self.default) and self.type == "text":
                     return f"{MYSQL_SQL_TYPES['varchar']}({self.length if self.length is not None else 255})"
                 return (
                     f"{MYSQL_SQL_TYPES[self.type]}({self.length})"
@@ -217,7 +217,7 @@ def sql_type(self, dialect: str):
 
         elif dialect == "mysql":
             if self.type in MYSQL_SQL_TYPES:
-                if self.unique and self.type == "text":
+                if (self.unique or self.default) and self.type == "text":
                     return f"{MYSQL_SQL_TYPES['varchar']}({self.length if self.length is not None else 255})"
                 return (
                     f"{MYSQL_SQL_TYPES[self.type]}({self.length})"

dataloom/tests/mysql/test_create_tables_mysql.py

@@ -0,0 +1,119 @@
+class TestCreatingTableMysql:
+    def test_2_pk_error(self):
+        from dataloom import Column, PrimaryKeyColumn, Dataloom, TableColumn, Model
+        import pytest
+        from typing import Optional
+
+        mysql_loom = Dataloom(
+            dialect="mysql", database="hi", password="root", user="root"
+        )
+        conn = mysql_loom.connect()
+
+        class User(Model):
+            __tablename__: Optional[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)
+        assert (
+            str(exc_info.value)
+            == "You have defined many field as primary keys which is not allowed. Fields (`_id`, `id`) are primary keys."
+        )
+        conn.close()
+
+    def test_no_pk_error(self):
+        import pytest
+        from dataloom import Model, Dataloom, Column, TableColumn
+        from typing import Optional
+
+        mysql_loom = Dataloom(
+            dialect="mysql", database="hi", password="root", user="root"
+        )
+        conn = mysql_loom.connect()
+
+        class User(Model):
+            __tablename__: Optional[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)
+        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
+
+        mysql_loom = Dataloom(
+            dialect="mysql", database="hi", password="root", user="root"
+        )
+        conn = mysql_loom.connect()
+
+        class Posts(Model):
+            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")
+            username = Column(type="text", nullable=False, default="Hello there!!")
+            name = Column(type="varchar", unique=True, length=255)
+
+        assert User._get_table_name() == "users"
+        assert Posts._get_table_name() == "posts"
+        conn.close()
+
+    def test_connect_sync(self):
+        from dataloom import Dataloom, Model, TableColumn, Column, PrimaryKeyColumn
+        from typing import Optional
+
+        class User(Model):
+            __tablename__: Optional[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")
+
+            id = PrimaryKeyColumn(type="int", nullable=False, auto_increment=True)
+            title = Column(type="text", nullable=False)
+
+        mysql_loom = Dataloom(
+            dialect="mysql", database="hi", password="root", user="root"
+        )
+        conn, tables = mysql_loom.connect_and_sync([User, Post], drop=True, force=True)
+        assert len(tables) == 2
+        assert sorted(tables) == sorted(["users", "posts"])
+
+        conn.close()
+
+    def test_syncing_tables(self):
+        from dataloom import Model, Dataloom, Column, PrimaryKeyColumn, TableColumn
+        from typing import Optional
+
+        mysql_loom = Dataloom(
+            dialect="mysql", database="hi", password="root", user="root"
+        )
+        conn = mysql_loom.connect()
+
+        class Post(Model):
+            __tablename__: Optional[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="int", auto_increment=True)
+            username = Column(type="text", nullable=False, default="Hello there!!")
+            name = Column(type="varchar", unique=True, length=255)
+
+        tables = mysql_loom.sync([User, Post], drop=True, force=True)
+        assert len(tables) == 2
+        assert sorted(tables) == sorted(["users", "posts"])
+        conn.close()