Forward all connection string params to py-core in bulkcopy

Author: bewithgaurav

mssql_python/cursor.py

@@ -17,7 +17,7 @@
 import warnings
 from typing import List, Union, Any, Optional, Tuple, Sequence, TYPE_CHECKING, Iterable
 from mssql_python.constants import ConstantsDDBC as ddbc_sql_const, SQLTypes
-from mssql_python.helpers import check_error
+from mssql_python.helpers import check_error, connstr_to_pycore_params
 from mssql_python.logging import logger
 from mssql_python import ddbc_bindings
 from mssql_python.exceptions import (
@@ -2451,6 +2451,7 @@ def nextset(self) -> Union[bool, None]:
         )
         return True
 
+    # ── Mapping from ODBC connection-string keywords (lowercase, as _parse returns)
     def _bulkcopy(
         self,
         table_name: str,
@@ -2585,38 +2586,10 @@ def _bulkcopy(
                 "Specify the target database explicitly to avoid accidentally writing to system databases."
             )
 
-        # Build connection context for bulk copy library
-        # Note: Password is extracted separately to avoid storing it in the main context
-        # dict that could be accidentally logged or exposed in error messages.
-        trust_cert = params.get("trustservercertificate", "yes").lower() in ("yes", "true")
-
-        # Parse encryption setting from connection string
-        encrypt_param = params.get("encrypt")
-        if encrypt_param is not None:
-            encrypt_value = encrypt_param.strip().lower()
-            if encrypt_value in ("yes", "true", "mandatory", "required"):
-                encryption = "Required"
-            elif encrypt_value in ("no", "false", "optional"):
-                encryption = "Optional"
-            else:
-                # Pass through unrecognized values (e.g., "Strict") to the underlying driver
-                encryption = encrypt_param
-        else:
-            encryption = "Optional"
-
-        context = {
-            "server": params.get("server"),
-            "database": params.get("database"),
-            "trust_server_certificate": trust_cert,
-            "encryption": encryption,
-        }
-
-        # Build pycore_context with appropriate authentication.
-        # For Azure AD: acquire a FRESH token right now instead of reusing
-        # the one from connect() time — avoids expired-token errors when
-        # bulkcopy() is called long after the original connection.
-        pycore_context = dict(context)
+        # Translate parsed connection string into the dict py-core expects.
+        pycore_context = connstr_to_pycore_params(params)
 
+        # Token acquisition — only thing cursor must handle (needs azure-identity SDK)
         if self.connection._auth_type:
             # Fresh token acquisition for mssql-py-core connection
             from mssql_python.auth import AADAuth
@@ -2633,10 +2606,6 @@ def _bulkcopy(
                 "Bulk copy: acquired fresh Azure AD token for auth_type=%s",
                 self.connection._auth_type,
             )
-        else:
-            # SQL Server authentication — use uid/password from connection string
-            pycore_context["user_name"] = params.get("uid", "")
-            pycore_context["password"] = params.get("pwd", "")
 
         pycore_connection = None
         pycore_cursor = None
@@ -2675,9 +2644,8 @@ def _bulkcopy(
         finally:
             # Clear sensitive data to minimize memory exposure
             if pycore_context:
-                pycore_context.pop("password", None)
-                pycore_context.pop("user_name", None)
-                pycore_context.pop("access_token", None)
+                for key in ("password", "user_name", "access_token"):
+                    pycore_context.pop(key, None)
             # Clean up bulk copy resources
             for resource in (pycore_cursor, pycore_connection):
                 if resource and hasattr(resource, "close"):

mssql_python/helpers.py

@@ -250,6 +250,82 @@ def _sanitize_for_logging(input_val: Any, max_length: int = max_log_length) -> s
     return True, None, sanitized_attr, sanitized_val
 
 
+def connstr_to_pycore_params(params: dict) -> dict:
+    """Translate parsed connection-string params into the dict that py-core's
+    ``connection.rs`` expects.
+
+    *params* uses the lowercase keys that ``_ConnectionStringParser._parse``
+    returns.  This function maps them to py-core's snake_case keys and
+    converts str values to bool/int where the Rust side expects native types.
+    Keys not recognised by py-core are silently dropped.
+    """
+    # connstr key (lowercase) → py-core dict key
+    key_map = {
+        # auth / credentials
+        "uid": "user_name",
+        "pwd": "password",
+        "trusted_connection": "trusted_connection",
+        "authentication": "authentication",
+        # server (accept parser synonyms)
+        "server": "server",
+        "addr": "server",
+        "address": "server",
+        # database
+        "database": "database",
+        "app": "application_name",
+        "applicationintent": "application_intent",
+        "workstationid": "workstation_id",
+        "language": "language",
+        # encryption / TLS (include snake_case alias the parser may emit)
+        "encrypt": "encryption",
+        "trustservercertificate": "trust_server_certificate",
+        "trust_server_certificate": "trust_server_certificate",
+        "hostnameincertificate": "host_name_in_certificate",
+        "servercertificate": "server_certificate",
+        # Kerberos
+        "serverspn": "server_spn",
+        # network
+        "multisubnetfailover": "multi_subnet_failover",
+        "ipaddresspreference": "ip_address_preference",
+        "keepalive": "keep_alive",
+        "keepaliveinterval": "keep_alive_interval",
+        # sizing / limits ("packet size" with space is a common pyodbc-ism)
+        "packetsize": "packet_size",
+        "packet size": "packet_size",
+        "connect_timeout": "connect_timeout",
+        "connectretrycount": "connect_retry_count",
+        "connectretryinterval": "connect_retry_interval",
+        # MARS
+        "mars_connection": "mars_enabled",
+    }
+    bool_keys = {"trust_server_certificate", "multi_subnet_failover", "mars_enabled"}
+    int_keys = {
+        "connect_timeout",
+        "packet_size",
+        "connect_retry_count",
+        "connect_retry_interval",
+        "keep_alive",
+        "keep_alive_interval",
+    }
+
+    ctx: dict = {}
+    for src, dst in key_map.items():
+        val = params.get(src)
+        if val is None:
+            continue
+        if dst in bool_keys:
+            ctx[dst] = val.lower() in ("yes", "true", "1") if isinstance(val, str) else bool(val)
+        elif dst in int_keys:
+            try:
+                ctx[dst] = int(val)
+            except (ValueError, TypeError):
+                pass  # let py-core use its default
+        else:
+            ctx[dst] = val
+
+    return ctx
+
+
 # Settings functionality moved here to avoid circular imports
 
 # Initialize the locale setting only once at module import time