connecting using connection-uri's

Changelog.md

@@ -1,3 +1,19 @@
+===
+Dataloom **`2.1.0`**
+===
+
+### Release Notes - `dataloom`
+
+We have release the new `dataloom` Version `2.1.0` (`2024-02-12`)
+
+##### Features
+
+- Connecting to databases using connection `uri` for all the supported dialects
+
+  ```py
+
+  ```
+
 ===
 Dataloom **`2.0.0`**
 ===

dataloom/conn/__init__.py

@@ -4,7 +4,7 @@
 from os import PathLike
 from typing_extensions import TypeAlias
 from typing import Optional
-
+from urllib.parse import urlparse, parse_qs
 from dataloom.constants import instances
 
 
@@ -87,3 +87,21 @@ def get_connection_options(**kwargs):
             raise UnsupportedDialectException(
                 "The dialect passed is not supported the supported dialects are: {'postgres', 'mysql', 'sqlite'}"
             )
+
+    @staticmethod
+    def get_mysql_uri_connection_options(uri: str) -> dict:
+        components = urlparse(uri)
+        user = components.username
+        password = components.password
+        hostname = components.hostname
+        port = components.port
+        db = components.path.lstrip("/")
+
+        return {
+            "user": user,
+            "password": password,
+            "host": hostname,
+            "port": port,
+            "database": db,
+            **parse_qs(components.query),
+        }

dataloom/keys.py

@@ -1,3 +1,6 @@
+# Configuration file for unit testing.
+
+
 push = True
 
 
@@ -6,18 +9,41 @@ class PgConfig:
         password = "postgres"
         database = "postgres"
         user = "postgres"
+        host = "0.0.0.0"
+        port = 5432
+        db = "hi"
+        connection_uri = f"postgresql://{user}:{password}@{host}:{port}/{db}"
     else:
         database = "postgres"
         user = "postgres"
         password = "root"
+        host = "localhost"
+        port = 5432
+        db = "hi"
+        connection_uri = f"postgresql://{user}:{password}@{host}:{port}/{db}"
 
 
 class MySQLConfig:
     if push:
         password = "testrootpass"
         database = "testdb"
         user = "root"
+        host = "0.0.0.0"
+        port = 3306
+        db = "hi"
+        connection_uri = f"mysql://{user}:{password}@{host}:{port}/{db}"
     else:
         database = "hi"
         user = "root"
         password = "root"
+        host = "localhost"
+        port = 3306
+        db = "hi"
+        connection_uri = f"mysql://{user}:{password}@{host}:{port}/{db}"
+
+
+class SQLiteConfig:
+    if push:
+        connection_uri = "sqlite:///database.db"
+    else:
+        connection_uri = r"C:\\Users\\RGaridzirai\\OneDrive - Walter Sisulu University\Desktop\\TINASHE CRISPEN G\\orm\\dataloom-py\\hi.db"

dataloom/loom/__init__.py

@@ -69,8 +69,9 @@ class Loom(ILoom):
 
     def __init__(
         self,
-        database: str,
         dialect: DIALECT_LITERAL,
+        connection_uri: Optional[str] = None,
+        database: Optional[str] = None,
         user: Optional[str] = None,
         host: Optional[str] = None,
         port: Optional[int] = None,
@@ -83,6 +84,7 @@ def __init__(
         self.sql_logger = sql_logger
         self.dialect = dialect
         self.logs_filename = logs_filename
+        self.connection_uri = connection_uri
 
         try:
             config = instances[dialect]
@@ -1194,27 +1196,41 @@ def connect(
 
         """
         if self.dialect == "postgres":
-            options = ConnectionOptionsFactory.get_connection_options(
-                **self.connection_options
-            )
-            with psycopg2.connect(**options) as conn:
-                self.conn = conn
+            if self.connection_uri is None:
+                options = ConnectionOptionsFactory.get_connection_options(
+                    **self.connection_options
+                )
+                with psycopg2.connect(**options) as conn:
+                    self.conn = conn
+            else:
+                with psycopg2.connect(self.connection_uri) as conn:
+                    self.conn = conn
         elif self.dialect == "mysql":
-            options = ConnectionOptionsFactory.get_connection_options(
-                **self.connection_options
+            options = (
+                ConnectionOptionsFactory.get_connection_options(
+                    **self.connection_options
+                )
+                if self.connection_uri is None
+                else ConnectionOptionsFactory.get_mysql_uri_connection_options(
+                    self.connection_uri
+                )
             )
             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
+            if self.connection_uri is None:
+                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:
-                with sqlite3.connect(**options) as conn:
+                with sqlite3.connect(self.connection_uri) as conn:
                     self.conn = conn
         else:
             raise UnsupportedDialectException(
@@ -1285,34 +1301,8 @@ def connect_and_sync(
         ...     conn.close()
 
         """
-
         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'}"
-                )
+            self.conn = self.connect()
             self.sql_obj = SQL(
                 conn=self.conn,
                 dialect=self.dialect,

dataloom/tests/mysql/test_connection_mysql.py → dataloom/tests/mysql/test_connection_options_mysql.py

@@ -2,7 +2,7 @@
 from mysql import connector
 
 
-class TestConnectionMySQL:
+class TestConnectionOptionsMySQL:
     def test_connect_with_non_existing_database(self):
         from dataloom import Loom
         from dataloom.keys import MySQLConfig

dataloom/tests/mysql/test_connection_uri_mysql.py

@@ -0,0 +1,81 @@
+import pytest
+from mysql import connector
+
+
+class TestConnectionURIMySQL:
+    def test_connect_with_non_existing_database(self):
+        from dataloom import Loom
+        from dataloom.keys import MySQLConfig
+
+        mysql_loom = Loom(
+            dialect="mysql",
+            connection_uri=f"mysql://{MySQLConfig.user}:{MySQLConfig.password}@{MySQLConfig.host}:{MySQLConfig.port}/non-exists",
+        )
+        with pytest.raises(connector.errors.ProgrammingError) as exc_info:
+            conn = mysql_loom.connect()
+            conn.close()
+        assert exc_info.value.msg == "Unknown database 'non-exists'"
+        assert exc_info.value.errno == 1049
+
+    def test_connect_with_wrong_password(self):
+        from dataloom import Loom
+        from dataloom.keys import MySQLConfig
+
+        mysql_loom = Loom(
+            dialect="mysql",
+            connection_uri=f"mysql://{MySQLConfig.user}:{MySQLConfig.password+'me'}@{MySQLConfig.host}:{MySQLConfig.port}/hi",
+        )
+        with pytest.raises(connector.errors.ProgrammingError) as exc_info:
+            conn = mysql_loom.connect()
+            conn.close()
+
+        assert str(exc_info.value.msg).startswith(
+            "Access denied for user 'root'@'"
+        ) and str(exc_info.value.msg).endswith("(using password: YES)")
+        assert exc_info.value.errno == 1045
+
+    def test_connect_with_wrong_user(self):
+        from dataloom import Loom
+        from dataloom.keys import MySQLConfig
+
+        mysql_loom = Loom(
+            dialect="mysql",
+            connection_uri=f"mysql://hey:{MySQLConfig.password}@{MySQLConfig.host}:{MySQLConfig.port}/hi",
+        )
+        with pytest.raises(connector.errors.ProgrammingError) as exc_info:
+            conn = mysql_loom.connect()
+            conn.close()
+        assert str(exc_info.value.msg).startswith(
+            "Access denied for user 'hey'@"
+        ) and str(exc_info.value.msg).endswith("(using password: YES)")
+        assert exc_info.value.errno == 1045
+
+    def test_connect_with_wrong_dialect(self):
+        from dataloom import Loom
+        from dataloom.keys import MySQLConfig
+        from dataloom.exceptions import UnsupportedDialectException
+
+        with pytest.raises(UnsupportedDialectException) as exc_info:
+            mysql_loom = Loom(
+                dialect="dialect",
+                connection_uri=f"mysql://{MySQLConfig.user}:{MySQLConfig.password}@{MySQLConfig.host}:{MySQLConfig.port}/hi",
+            )
+            conn = mysql_loom.connect()
+            conn.close()
+
+        assert (
+            str(exc_info.value)
+            == "The dialect passed is not supported the supported dialects are: {'postgres', 'mysql', 'sqlite'}"
+        )
+
+    def test_connect_correct_connection(self):
+        from dataloom import Loom
+        from dataloom.keys import MySQLConfig
+
+        mysql_loom = Loom(
+            dialect="mysql",
+            connection_uri=f"mysql://{MySQLConfig.user}:{MySQLConfig.password}@{MySQLConfig.host}:{MySQLConfig.port}/hi",
+        )
+        conn = mysql_loom.connect()
+        conn.close()
+        assert conn is not None

dataloom/tests/postgres/test_connection_pg.py → dataloom/tests/postgres/test_connection_options_pg.py

@@ -1,7 +1,7 @@
 import pytest
 
 
-class TestConnectionPG:
+class TestConnectionOptionsPG:
     def test_connect_with_non_existing_database(self):
         from dataloom import Loom
         from dataloom.keys import PgConfig

dataloom/tests/postgres/test_connection_uri_pg.py

@@ -0,0 +1,83 @@
+import pytest
+
+
+class TestConnectionURIPG:
+    def test_connect_with_non_existing_database(self):
+        from dataloom import Loom
+        from dataloom.keys import PgConfig
+
+        pg_loom = Loom(
+            dialect="postgres",
+            connection_uri=f"postgresql://{PgConfig.user}:{PgConfig.password}@{PgConfig.host}:{PgConfig.port}/mew",
+        )
+        with pytest.raises(Exception) as exc_info:
+            conn = pg_loom.connect()
+            conn.close()
+        assert (
+            str(exc_info.value.args[0]).strip()
+            == 'connection to server at "localhost" (::1), port 5432 failed: FATAL:  database "mew" does not exist'
+        )
+
+    def test_connect_with_wrong_password(self):
+        from dataloom import Loom
+        from dataloom.keys import PgConfig
+
+        pg_loom = Loom(
+            dialect="postgres",
+            connection_uri=f"postgresql://{PgConfig.user}:{PgConfig.password+'-'}@{PgConfig.host}:{PgConfig.port}/hi",
+        )
+        with pytest.raises(Exception) as exc_info:
+            conn = pg_loom.connect()
+            conn.close()
+
+        assert (
+            str(exc_info.value.args[0]).strip()
+            == 'connection to server at "localhost" (::1), port 5432 failed: FATAL:  password authentication failed for user "postgres"'
+        )
+
+    def test_connect_with_wrong_user(self):
+        from dataloom import Loom
+        from dataloom.keys import PgConfig
+
+        pg_loom = Loom(
+            dialect="postgres",
+            connection_uri=f"postgresql://postgre-u:{PgConfig.password}@{PgConfig.host}:{PgConfig.port}/hi",
+        )
+        with pytest.raises(Exception) as exc_info:
+            conn = pg_loom.connect()
+            conn.close()
+
+        assert (
+            str(exc_info.value.args[0]).strip()
+            == 'connection to server at "localhost" (::1), port 5432 failed: FATAL:  password authentication failed for user "postgre-u"'
+        )
+
+    def test_connect_with_wrong_dialect(self):
+        from dataloom import Loom
+        from dataloom.exceptions import UnsupportedDialectException
+        from dataloom.keys import PgConfig
+
+        with pytest.raises(UnsupportedDialectException) as exc_info:
+            pg_loom = Loom(
+                dialect="peew",
+                connection_uri=f"postgresql://{PgConfig.user}:{PgConfig.password}@{PgConfig.host}:{PgConfig.port}/hi",
+            )
+            conn = pg_loom.connect()
+            conn.close()
+        assert (
+            str(exc_info.value)
+            == "The dialect passed is not supported the supported dialects are: {'postgres', 'mysql', 'sqlite'}"
+        )
+
+    def test_connect_correct_connection(self):
+        from dataloom import Loom
+        from dataloom.keys import PgConfig
+
+        pg_loom = Loom(
+            dialect="postgres",
+            connection_uri=f"postgresql://{PgConfig.user}:{PgConfig.password}@{PgConfig.host}:{PgConfig.port}/hi",
+        )
+        conn = pg_loom.connect()
+        conn.close()
+
+        assert conn is not None