Merge pull request #145 from jacobshaw42/master

modified SqlCreds.from_engine to allow for engine from non-encoded url

bcpandas/main.py

@@ -137,20 +137,30 @@ def from_engine(cls, engine: sa.engine.base.Engine) -> "SqlCreds":
         """
         try:
             # get the odbc url part from the engine, split by ';' delimiter
-            conn_url = engine.url.query["odbc_connect"].split(";")
-            # convert into dict
-            conn_dict = {x.split("=")[0].lower(): x.split("=")[1] for x in conn_url if "=" in x}
-
-            if "," in conn_dict["server"]:
-                conn_dict["port"] = int(conn_dict["server"].split(",")[1])
-
-            sql_creds = cls(
-                server=conn_dict["server"].replace("tcp:", "").split(",")[0],
-                database=conn_dict["database"],
-                username=conn_dict.get("uid", None),
-                password=conn_dict.get("pwd", None),
-                port=conn_dict.get("port", None),
-            )
+            if "odbc_connect" in engine.url.query:
+                conn_url = engine.url.query["odbc_connect"].split(";")
+                # convert into dict
+                conn_dict = {
+                    x.split("=")[0].lower(): x.split("=")[1] for x in conn_url if "=" in x
+                }
+
+                if "," in conn_dict["server"]:
+                    conn_dict["port"] = int(conn_dict["server"].split(",")[1])
+                sql_creds = cls(
+                    server=conn_dict["server"].replace("tcp:", "").split(",")[0],
+                    database=conn_dict["database"],
+                    username=conn_dict.get("uid", None),
+                    password=conn_dict.get("pwd", None),
+                    port=conn_dict.get("port", None),
+                )
+            elif "driver" in engine.url.query:
+                sql_creds = cls(
+                    server=engine.url.host,
+                    database=engine.url.database,
+                    username=engine.url.username,
+                    password=engine.url.password,
+                    port=engine.url.port,
+                )
 
             # add Engine object as attribute
             sql_creds.engine = engine

tests/test_sqlcreds.py

@@ -250,6 +250,28 @@ def test_sql_creds_from_sqlalchemy():
     )
 
 
+def test_sql_creds_from_sqlalchemy_no_url_encoding():
+    """
+    Tests that the SqlCreds object can be created from a SqlAlchemy engine using a traditional url
+    (no odbc_connect parameter)
+
+    * With Username and Password
+    * Use default port (1433)
+    """
+    url = "mssql+pyodbc://test_user:test_password@test_server:1433/test_database?driver=ODBC+Driver+99+for+SQL+Server"
+    mssql_engine = create_engine(url)
+    creds = SqlCreds.from_engine(mssql_engine)
+    assert creds.server == "test_server"
+    assert creds.database == "test_database"
+    assert creds.username == "test_user"
+    assert creds.password == "test_password"
+    assert creds.port == 1433
+    assert creds.with_krb_auth is False
+    assert isinstance(creds.engine, engine.Connectable)
+    assert creds.engine == mssql_engine
+    assert creds.engine.url == mssql_engine.url
+
+
 def test_sql_creds_from_sqlalchemy_windows_auth():
     """
     Tests that the SqlCreds object can be created from a SqlAlchemy engine