connections-objects

dataloom/__init__.py

@@ -0,0 +1,3 @@
+from dataloom.db import Dataloom
+
+dataloom = Dataloom()

dataloom/conn/__init__.py

@@ -0,0 +1,83 @@
+from dataclasses import dataclass, field
+from dataloom.exceptions import UnsupportedDialect
+from sqlite3.dbapi2 import Connection
+from os import PathLike
+from typing_extensions import TypeAlias
+
+StrOrBytesPath: TypeAlias = str | bytes | PathLike[str] | PathLike[bytes]
+
+
+@dataclass(kw_only=True)
+class Dialect:
+    def connection_options[T](self) -> T:
+        return {}
+
+
+@dataclass(kw_only=True)
+class PostgresDialect(Dialect):
+    # "postgres://postgres:postgres@localhost:5432/db"
+    connection_string: str | None = field(default=None)
+    database: str
+    user: str | None = field(default="postgres")
+    host: str | None = field(default="localhost")
+    port: int | None = field(default=5432)
+    password: str | None = field(default="postgres")
+
+    def connection_options[T](self) -> T:
+        return (
+            self.connection_string if self.connection_string is not None else vars(self)
+        )
+
+
+@dataclass(kw_only=True)
+class MySQLDialect(Dialect):
+    connection_string: str | None = field(default=None)
+    database: str
+    user: str | None = field(default="root")
+    host: str | None = field(default="localhost")
+    port: int | None = field(default=3306)
+    password: str | None = field(default="root")
+
+    def connection_options[T](self) -> T:
+        return (
+            self.connection_string if self.connection_string is not None else vars(self)
+        )
+
+
+@dataclass(kw_only=True)
+class SQLiteDialect(Dialect):
+    database: StrOrBytesPath = field(default="users.db")
+    timeout: float | None = field(default=None)
+    detect_types: int | None = field(default=None)
+    isolation_level: str | None = field(default=None)
+    check_same_thread: bool = field(default=None)
+    factory: type[Connection] | None = field(default=None)
+    cached_statements: int | None = field(default=None)
+    uri: bool | None = field(default=None)
+
+    def connection_options[T](self) -> T:
+        return vars(self)
+
+
+@dataclass
+class ConnectionOptionsFactory:
+    @staticmethod
+    def get_connection_options(dialect, **kwargs):
+        if dialect == "postgres":
+            return {
+                k: v
+                for k, v in PostgresDialect(**kwargs).connection_options().items()
+                if v is not None
+            }
+        elif dialect == "sqlite":
+            return SQLiteDialect(**kwargs).connection_options()
+        elif dialect == "mysql":
+            return {
+                k: v
+                for k, v in MySQLDialect(**kwargs).connection_options().items()
+                if v is not None
+            }
+        else:
+            raise UnsupportedDialect(
+                "The dialect passed is not supported the supported dialects are: {'postgres', 'mysql', 'sqlite'}"
+            )

orm/db/__init__.py → dataloom/db/__init__.py

@@ -1,9 +1,13 @@
 import psycopg2
+from mysql import connector
+import sqlite3
 import inspect
-from orm.constants import instances
-from orm.model.statements import Statements
-from orm.model.model import Model
-from orm.model.column import (
+from dataloom.constants import instances
+from dataloom.model.statements import Statements
+from dataloom.exceptions import UnsupportedDialect
+from dataloom.model.model import Model
+from dataloom.conn import ConnectionOptionsFactory
+from dataloom.model.column import (
     Column,
     UpdatedAtColumn,
     PrimaryKeyColumn,
@@ -12,6 +16,32 @@
 )
 
 
+class Dataloom:
+    conn = None
+
+    def connect(self, dialect, **kwargs):
+        if dialect == "postgres":
+            options = ConnectionOptionsFactory.get_connection_options(dialect, **kwargs)
+            self.conn = psycopg2.connect(**options)
+            return self.conn
+        elif dialect == "mysql":
+            options = ConnectionOptionsFactory.get_connection_options(dialect, **kwargs)
+            self.conn = connector.connect(**options)
+            return self.conn
+        elif dialect == "sqlite":
+            options = ConnectionOptionsFactory.get_connection_options(dialect, **kwargs)
+            self.conn = (
+                sqlite3.connect(options.get("database"))
+                if "database" in options
+                else sqlite3.connect(**options)
+            )
+            return self.conn
+        else:
+            raise UnsupportedDialect(
+                "The dialect passed is not supported the supported dialects are: {'postgres', 'mysql', 'sqlite'}"
+            )
+
+
 class Database:
     def __init__(
         self,

orm/exceptions/__init__.py → dataloom/exceptions/__init__.py

@@ -4,3 +4,7 @@ class PkNotDefinedException(Exception):
 
 class TooManyPkException(Exception):
     pass
+
+
+class UnsupportedDialect(ValueError):
+    pass

orm/model/column.py → dataloom/model/column.py

@@ -1,4 +1,4 @@
-from orm.types import POSTGRES_SQL_TYPES
+from dataloom.types import POSTGRES_SQL_TYPES
 
 
 class CreatedAtColumn:

orm/model/model.py → dataloom/model/model.py

@@ -1,13 +1,13 @@
 from typing import Any
-from orm.model.column import (
+from dataloom.model.column import (
     Column,
     CreatedAtColumn,
     UpdatedAtColumn,
     ForeignKeyColumn,
     PrimaryKeyColumn,
 )
-from orm.model.statements import Statements
-from orm.exceptions import *
+from dataloom.model.statements import Statements
+from dataloom.exceptions import *
 import inspect
 from datetime import datetime
 import re
@@ -388,7 +388,6 @@ def _get_update_bulk_where_stm(cls, filters: dict = {}, args: dict = {}):
             placeholder_values.append(f'"{updatedAtColumName}" = %s')
             values.append(current_time_stamp)
 
-
         sql = Statements.UPDATE_BULK_WHERE_COMMAND.format(
             table_name=cls._get_name(),
             placeholder_values=", ".join(placeholder_values),

orm/tests/test_connection.py → dataloom/tests/test_connection.py

@@ -1,18 +1,18 @@
 class TestConnectionPG:
     def test_connect(self):
-        from orm.db import Database
-        from orm.keys import password, database, user
+        from dataloom.db import Database
+        from dataloom.keys import password, database, user
 
         db = Database(database, password=password, user=user)
         conn = db.connect()
         assert conn.status == 1
         conn.close()
 
     def test_connect_sync(self):
-        from orm.db import Database
-        from orm.keys import password, database, user
-        from orm.model.model import Model
-        from orm.model.column import Column, PrimaryKeyColumn
+        from dataloom.db import Database
+        from dataloom.keys import password, database, user
+        from dataloom.model.model import Model
+        from dataloom.model.column import Column, PrimaryKeyColumn
 
         class User(Model):
             __tablename__ = "users"

orm/tests/test_create_table.py → dataloom/tests/test_create_table.py

@@ -1,9 +1,9 @@
 class TestCreatingTablePG:
     def test_2_pk_error(self):
-        from orm.db import Database
-        from orm.model.column import Column, PrimaryKeyColumn
-        from orm.model.model import Model
-        from orm.keys import password, database, user
+        from dataloom.db import Database
+        from dataloom.model.column import Column, PrimaryKeyColumn
+        from dataloom.model.model import Model
+        from dataloom.keys import password, database, user
         import pytest
 
         db = Database(database, password=password, user=user)
@@ -26,10 +26,10 @@ class User(Model):
         conn.close()
 
     def test_no_pk_error(self):
-        from orm.db import Database
-        from orm.model.column import Column
-        from orm.model.model import Model
-        from orm.keys import password, database, user
+        from dataloom.db import Database
+        from dataloom.model.column import Column
+        from dataloom.model.model import Model
+        from dataloom.keys import password, database, user
         import pytest
 
         db = Database(database, password=password, user=user)
@@ -47,10 +47,10 @@ class User(Model):
         conn.close()
 
     def test_table_name(self):
-        from orm.db import Database
-        from orm.model.column import Column, PrimaryKeyColumn
-        from orm.model.model import Model
-        from orm.keys import database, password, user
+        from dataloom.db import Database
+        from dataloom.model.column import Column, PrimaryKeyColumn
+        from dataloom.model.model import Model
+        from dataloom.keys import database, password, user
 
         db = Database(database, password=password, user=user)
         conn = db.connect()
@@ -70,10 +70,10 @@ class User(Model):
         conn.close()
 
     def test_connect_sync(self):
-        from orm.db import Database
-        from orm.keys import password, database, user
-        from orm.model.model import Model
-        from orm.model.column import Column, PrimaryKeyColumn
+        from dataloom.db import Database
+        from dataloom.keys import password, database, user
+        from dataloom.model.model import Model
+        from dataloom.model.column import Column, PrimaryKeyColumn
 
         class User(Model):
             __tablename__ = "users"
@@ -97,10 +97,10 @@ class Post(Model):
         conn.close()
 
     def test_syncing_tables(self):
-        from orm.db import Database
-        from orm.keys import password, database, user
-        from orm.model.model import Model
-        from orm.model.column import Column, PrimaryKeyColumn
+        from dataloom.db import Database
+        from dataloom.keys import password, database, user
+        from dataloom.model.model import Model
+        from dataloom.model.column import Column, PrimaryKeyColumn
 
         class User(Model):
             __tablename__ = "users"

orm/tests/test_delete.py → dataloom/tests/test_delete.py

@@ -1,9 +1,9 @@
 class TestDeletingOnPG:
     def test_delete_by_pk_single_fn(self):
-        from orm.db import Database
-        from orm.model.column import Column
-        from orm.model.model import Model, PrimaryKeyColumn
-        from orm.keys import password, database, user
+        from dataloom.db import Database
+        from dataloom.model.column import Column
+        from dataloom.model.model import Model, PrimaryKeyColumn
+        from dataloom.keys import password, database, user
 
         db = Database(database, password=password, user=user)
         conn = db.connect()
@@ -25,10 +25,10 @@ class User(Model):
         conn.close()
 
     def test_delete_one_fn(self):
-        from orm.db import Database
-        from orm.model.column import Column
-        from orm.model.model import Model, PrimaryKeyColumn
-        from orm.keys import password, database, user
+        from dataloom.db import Database
+        from dataloom.model.column import Column
+        from dataloom.model.model import Model, PrimaryKeyColumn
+        from dataloom.keys import password, database, user
 
         db = Database(database, password=password, user=user)
         conn = db.connect()
@@ -60,10 +60,10 @@ class User(Model):
         conn.close()
 
     def test_delete_bulk_fn(self):
-        from orm.db import Database
-        from orm.model.column import Column
-        from orm.model.model import Model, PrimaryKeyColumn
-        from orm.keys import password, database, user
+        from dataloom.db import Database
+        from dataloom.model.column import Column
+        from dataloom.model.model import Model, PrimaryKeyColumn
+        from dataloom.keys import password, database, user
 
         db = Database(database, password=password, user=user)
         conn = db.connect()

orm/tests/test_insert.py → dataloom/tests/test_insert.py

@@ -1,9 +1,9 @@
 class TestInsertingOnPG:
     def test_insetting_single_document(self):
-        from orm.db import Database
-        from orm.model.column import Column
-        from orm.model.model import Model, PrimaryKeyColumn
-        from orm.keys import password, database, user
+        from dataloom.db import Database
+        from dataloom.model.column import Column
+        from dataloom.model.model import Model, PrimaryKeyColumn
+        from dataloom.keys import password, database, user
 
         db = Database(database, password=password, user=user)
         conn = db.connect()
@@ -21,16 +21,16 @@ class Users(Model):
         conn.close()
 
     def test_insetting_multiple_document(self):
-        from orm.db import Database
-        from orm.model.column import Column
-        from orm.model.model import (
+        from dataloom.db import Database
+        from dataloom.model.column import Column
+        from dataloom.model.model import (
             Model,
             CreatedAtColumn,
             UpdatedAtColumn,
             ForeignKeyColumn,
             PrimaryKeyColumn,
         )
-        from orm.keys import password, database, user
+        from dataloom.keys import password, database, user
 
         db = Database(database, password=password, user=user)
 
@@ -65,16 +65,16 @@ class Post(Model):
         conn.close()
 
     def test_relational_instances(self):
-        from orm.db import Database
-        from orm.model.column import Column
-        from orm.model.model import (
+        from dataloom.db import Database
+        from dataloom.model.column import Column
+        from dataloom.model.model import (
             Model,
             CreatedAtColumn,
             UpdatedAtColumn,
             ForeignKeyColumn,
             PrimaryKeyColumn,
         )
-        from orm.keys import password, database, user
+        from dataloom.keys import password, database, user
 
         db = Database(database, password=password, user=user)
 

orm/tests/test_query.py → dataloom/tests/test_query.py

@@ -1,9 +1,9 @@
 class TestQueryingPG:
     def test_querying_data(self):
-        from orm.db import Database
-        from orm.model.column import Column, PrimaryKeyColumn
-        from orm.model.model import Model
-        from orm.keys import password, database, user
+        from dataloom.db import Database
+        from dataloom.model.column import Column, PrimaryKeyColumn
+        from dataloom.model.model import Model
+        from dataloom.keys import password, database, user
 
         db = Database(database, password=password, user=user)
         conn = db.connect()