propose solution for issues #130 #131 (#133)

bcpandas/main.py

@@ -71,6 +71,7 @@ def __init__(
         self.server = server
         self.database = database
         self.port = port
+        self.odbc_kwargs = odbc_kwargs
 
         if driver_version is None:
             all_drivers: List[str] = pyodbc.drivers()
@@ -79,8 +80,10 @@ def __init__(
             ]
             new_driver_version: int = max(int(v) for v in driver_candidates if v.isnumeric())
             self.driver = f"{{ODBC Driver {new_driver_version} for SQL Server}}"
+            self.driver_version = new_driver_version  # simplifies copy construction
         else:
             self.driver = f"{{ODBC Driver {driver_version} for SQL Server}}"
+            self.driver_version = driver_version  # simplifies copy construction
 
         # Append a comma for use in connection strings (optionally blank)
         if port:
@@ -95,7 +98,7 @@ def __init__(
             self.username = username
             self.password = password
             self.with_krb_auth = False
-            db_url += f"UID={username};PWD={password}"
+            db_url += f"UID={username};PWD={password};"
         else:
             self.username = ""
             self.password = ""

bcpandas/utils.py

@@ -5,6 +5,7 @@
 """
 
 import logging
+import sys
 from pathlib import Path
 import random
 import shlex
@@ -70,6 +71,12 @@ def bcp(
         auth = ["-T"]
     else:
         auth = ["-U", quote_this(creds.username), "-P", quote_this(creds.password)]
+    if creds.odbc_kwargs:
+        kwargs = {k.lower(): v for k, v in creds.odbc_kwargs.items()}
+        false_values = ("n", "no", "f", "false", "off", "0")
+
+        if sys.platform != "win32" and "encrypt" in kwargs:
+            auth += [f"-Y{'m' if kwargs['encrypt'] not in false_values else 'o'}"]
 
     # prepare SQL item string
     if sql_type == QUERY:

tests/test_sqlcreds.py

@@ -68,7 +68,7 @@ def test_sql_creds_for_username_password():
     assert isinstance(creds.engine, engine.Connectable)
     assert str(creds.engine.url) == _quote_engine_url(
         "Driver={ODBC Driver 99 for SQL Server};Server=tcp:test_server,1433;Database=test_database;"
-        "UID=test_user;PWD=test_password"
+        "UID=test_user;PWD=test_password;"
     )
 
 
@@ -97,7 +97,7 @@ def test_sql_creds_for_username_password_version_not_specified():
     assert isinstance(creds.engine, engine.Connectable)
     assert (
         new_url
-        == "mssql+pyodbc:///?odbc_connect=Driver%D%BODBC+Driver++for+SQL+Server%D%BServer%Dtcp%Atest_server%C%BDatabase%Dtest_database%BUID%Dtest_user%BPWD%Dtest_password"
+        == "mssql+pyodbc:///?odbc_connect=Driver%D%BODBC+Driver++for+SQL+Server%D%BServer%Dtcp%Atest_server%C%BDatabase%Dtest_database%BUID%Dtest_user%BPWD%Dtest_password%B"
     )
 
 
@@ -140,7 +140,7 @@ def test_sql_creds_for_username_password_non_default_port():
     assert isinstance(creds.engine, engine.Connectable)
     assert str(creds.engine.url) == _quote_engine_url(
         "Driver={ODBC Driver 99 for SQL Server};Server=tcp:test_server,9999;Database=test_database;"
-        "UID=test_user;PWD=test_password"
+        "UID=test_user;PWD=test_password;"
     )
 
 
@@ -180,7 +180,7 @@ def test_sql_creds_for_username_password_blank_port():
     assert creds.port is None
     assert str(creds.engine.url) == _quote_engine_url(
         "Driver={ODBC Driver 99 for SQL Server};Server=tcp:test_server;Database=test_database;"
-        "UID=test_user;PWD=test_password"
+        "UID=test_user;PWD=test_password;"
     )
 
 
@@ -203,6 +203,26 @@ def test_sql_creds_for_windows_auth_blank_port():
     )
 
 
+def test_sql_creds_for_odbc_kwargs():
+    """
+    Tests that the SqlCreds object can be created
+
+    * Without Username and Password (Windows Auth)
+    * Use blank port
+    """
+    creds = SqlCreds(
+        server="test_server",
+        database="test_database",
+        username="me",
+        password="secret",
+        driver_version=99,
+        odbc_kwargs=dict(Encrypt="yes"),
+    )
+    assert str(creds.engine.url) == _quote_engine_url(
+        "Driver={ODBC Driver 99 for SQL Server};Server=tcp:test_server,1433;Database=test_database;UID=me;PWD=secret;Encrypt=yes"
+    )
+
+
 def test_sql_creds_from_sqlalchemy():
     """
     Tests that the SqlCreds object can be created from a SqlAlchemy engine
@@ -389,3 +409,31 @@ def test_sqlcreds_connection_from_sqlalchemy(sql_creds):
     df = pd.read_sql(con=test_engine, sql="SELECT TOP 1 * FROM sys.objects")
 
     assert df.shape[0] == 1
+
+
+@pytest.mark.usefixtures("database")
+def test_sql_creds_connection_for_odbc_kwargs_encrypt(sql_creds):
+    """
+    Tests that the SqlCreds object can be created
+
+    * Without Username and Password (Windows Auth)
+    * Use blank port
+    """
+    creds = SqlCreds(
+        server=sql_creds.server,
+        database=sql_creds.database,
+        username=sql_creds.username,
+        password=sql_creds.password,
+        driver_version=sql_creds.driver_version,
+        port=sql_creds.port,
+        odbc_kwargs=dict(Encrypt="no"),
+    )
+    assert str(creds.engine.url) == _quote_engine_url(
+        f"Driver={sql_creds.driver};Server=tcp:{sql_creds.server},{sql_creds.port};"
+        f"Database={sql_creds.database};UID={sql_creds.username};"
+        f"PWD={sql_creds.password};Encrypt=no"
+    )
+    # Check the SqlCreds version works
+    df = pd.read_sql(con=creds.engine, sql="SELECT TOP 1 * FROM sys.objects")
+
+    assert df.shape[0] == 1

tests/test_utils.py

@@ -1,3 +1,4 @@
+import sys
 from collections import namedtuple
 from pathlib import Path
 import tempfile
@@ -18,14 +19,15 @@ def fixture_run_cmd_capture(monkeypatch):
 
 
 def test_bcpandas_creates_command_without_port_if_default(run_cmd):
-    Creds = namedtuple("Creds", "server port database with_krb_auth username password")
+    Creds = namedtuple("Creds", "server port database with_krb_auth username password odbc_kwargs")
     creds = Creds(
         server="localhost",
         port=1433,
         database="DB",
         with_krb_auth=False,
         username="me",
         password="secret",
+        odbc_kwargs=None,
     )
     utils.bcp("table", "in", "", creds, True)
     assert run_cmd.call_args == mock.call(
@@ -49,14 +51,15 @@ def test_bcpandas_creates_command_without_port_if_default(run_cmd):
 
 
 def test_bcpandas_creates_command_with_port_if_not_default(run_cmd):
-    Creds = namedtuple("Creds", "server port database with_krb_auth username password")
+    Creds = namedtuple("Creds", "server port database with_krb_auth username password odbc_kwargs")
     creds = Creds(
         server="localhost",
         port=1234,
         database="DB",
         with_krb_auth=False,
         username="me",
         password="secret",
+        odbc_kwargs=None,
     )
     utils.bcp("table", "in", "", creds, True)
     assert run_cmd.call_args == mock.call(
@@ -79,6 +82,72 @@ def test_bcpandas_creates_command_with_port_if_not_default(run_cmd):
     )
 
 
+def test_bcpandas_creates_command_with_encrypt_no(run_cmd):
+    Creds = namedtuple("Creds", "server port database with_krb_auth username password odbc_kwargs")
+    creds = Creds(
+        server="localhost",
+        port=1433,
+        database="DB",
+        with_krb_auth=False,
+        username="me",
+        password="secret",
+        odbc_kwargs=dict(encrypt="no"),
+    )
+    utils.bcp("table", "in", "", creds, True)
+    assert run_cmd.call_args == mock.call(
+        [
+            "bcp",
+            "dbo.table",
+            "in",
+            "",
+            "-S",
+            "localhost",
+            "-d",
+            "DB",
+            "-q",
+            "-U",
+            "me",
+            "-P",
+            "secret",
+            "-Yo",
+        ],
+        print_output=True,
+    )
+
+
+def test_bcpandas_creates_command_with_encrypt_yes(run_cmd):
+    Creds = namedtuple("Creds", "server port database with_krb_auth username password odbc_kwargs")
+    creds = Creds(
+        server="localhost",
+        port=1433,
+        database="DB",
+        with_krb_auth=False,
+        username="me",
+        password="secret",
+        odbc_kwargs=dict(Encrypt="1"),
+    )
+    utils.bcp("table", "in", "", creds, True)
+    assert run_cmd.call_args == mock.call(
+        [
+            "bcp",
+            "dbo.table",
+            "in",
+            "",
+            "-S",
+            "localhost",
+            "-d",
+            "DB",
+            "-q",
+            "-U",
+            "me",
+            "-P",
+            "secret",
+        ]
+        + (["-Ym"] if sys.platform != "win32" else []),
+        print_output=True,
+    )
+
+
 @pytest.mark.usefixtures("database")
 def test_bcp_login_failure(sql_creds: SqlCreds):
     wrong_sql_creds = SqlCreds(