FEAT: Improved Connection String handling in Python

eng/pipelines/build-whl-pipeline.yml

@@ -340,7 +340,7 @@
         python -m pytest -v
       displayName: 'Run Pytest to validate bindings'
       env:
-        DB_CONNECTION_STRING: 'Driver=ODBC Driver 18 for SQL Server;Server=tcp:127.0.0.1,1433;Database=master;Uid=SA;Pwd=$(DB_PASSWORD);TrustServerCertificate=yes'
+        DB_CONNECTION_STRING: 'Server=tcp:127.0.0.1,1433;Database=master;Uid=SA;Pwd=$(DB_PASSWORD);TrustServerCertificate=yes'
 
     # Build wheel package for universal2
     - script: |
@@ -801,7 +801,7 @@
 
       displayName: 'Test wheel installation and basic functionality on $(BASE_IMAGE)'
       env:
-        DB_CONNECTION_STRING: 'Driver=ODBC Driver 18 for SQL Server;Server=localhost;Database=TestDB;Uid=SA;Pwd=$(DB_PASSWORD);TrustServerCertificate=yes'
+        DB_CONNECTION_STRING: 'Server=localhost;Database=TestDB;Uid=SA;Pwd=$(DB_PASSWORD);TrustServerCertificate=yes'
 
     # Run pytest with source code while testing installed wheel
     - script: |
@@ -856,7 +856,7 @@
         "
       displayName: 'Run pytest suite on $(BASE_IMAGE) $(ARCH)'
       env:
-        DB_CONNECTION_STRING: 'Driver=ODBC Driver 18 for SQL Server;Server=localhost;Database=TestDB;Uid=SA;Pwd=$(DB_PASSWORD);TrustServerCertificate=yes'
+        DB_CONNECTION_STRING: 'Server=localhost;Database=TestDB;Uid=SA;Pwd=$(DB_PASSWORD);TrustServerCertificate=yes'
       continueOnError: true  # Don't fail pipeline if tests fail
 
     # Cleanup

mssql_python/__init__.py

@@ -25,6 +25,7 @@
     InternalError,
     ProgrammingError,
     NotSupportedError,
+    ConnectionStringParseError,
 )
 
 # Type Objects
@@ -46,6 +47,10 @@
 # Connection Objects
 from .db_connection import connect, Connection
 
+# Connection String Handling
+from .connection_string_parser import _ConnectionStringParser
+from .connection_string_builder import _ConnectionStringBuilder
+
 # Cursor Objects
 from .cursor import Cursor
 

mssql_python/connection.py

@@ -41,6 +41,9 @@
 )
 from mssql_python.auth import process_connection_string
 from mssql_python.constants import ConstantsDDBC, GetInfoConstants
+from mssql_python.connection_string_parser import _ConnectionStringParser
+from mssql_python.connection_string_builder import _ConnectionStringBuilder
+from mssql_python.constants import _RESERVED_PARAMETERS
 
 if TYPE_CHECKING:
     from mssql_python.row import Row
@@ -242,39 +245,62 @@ def _construct_connection_string(
         self, connection_str: str = "", **kwargs: Any
     ) -> str:
         """
-        Construct the connection string by concatenating the connection string
-        with key/value pairs from kwargs.
-
+        Construct the connection string by parsing, validating, and merging parameters.
+
+        This method performs a 6-step process:
+        1. Parse and validate the base connection_str (validates against allowlist)
+        2. Normalize parameter names (e.g., addr/address -> Server, uid -> UID)
+        3. Merge kwargs (which override connection_str params after normalization)
+        4. Build connection string from normalized, merged params
+        5. Add Driver and APP parameters (always controlled by the driver)
+        6. Return the final connection string
+
         Args:
             connection_str (str): The base connection string.
             **kwargs: Additional key/value pairs for the connection string.
 
         Returns:
-            str: The constructed connection string.
+            str: The constructed and validated connection string.
         """
-        # Add the driver attribute to the connection string
-        conn_str = add_driver_to_connection_str(connection_str)
-
-        # Add additional key-value pairs to the connection string
+
+        # Step 1: Parse base connection string with allowlist validation
+        # The parser validates everything: unknown params, reserved params, duplicates, syntax
+        parser = _ConnectionStringParser(validate_keywords=True)
+        parsed_params = parser._parse(connection_str)
+
+        # Step 2: Normalize parameter names (e.g., addr/address -> Server, uid -> UID)
+        # This handles synonym mapping and deduplication via normalized keys
+        normalized_params = _ConnectionStringParser._normalize_params(parsed_params, warn_rejected=False)
+
+        # Step 3: Process kwargs and merge with normalized_params
+        # kwargs override connection string values (processed after, so they take precedence)
         for key, value in kwargs.items():
-            if key.lower() == "host" or key.lower() == "server":
-                key = "Server"
-            elif key.lower() == "user" or key.lower() == "uid":
-                key = "Uid"
-            elif key.lower() == "password" or key.lower() == "pwd":
-                key = "Pwd"
-            elif key.lower() == "database":
-                key = "Database"
-            elif key.lower() == "encrypt":
-                key = "Encrypt"
-            elif key.lower() == "trust_server_certificate":
-                key = "TrustServerCertificate"
+            normalized_key = _ConnectionStringParser.normalize_key(key)
+            if normalized_key:
+                # Driver and APP are reserved - raise error if user tries to set them
+                if normalized_key in _RESERVED_PARAMETERS:
+                    raise ValueError(
+                        f"Connection parameter '{key}' is reserved and controlled by the driver. "
+                        f"It cannot be set by the user."
+                    )
+                # kwargs override any existing values from connection string
+                normalized_params[normalized_key] = str(value)
             else:
-                continue
-            conn_str += f"{key}={value};"
-
-        log("info", "Final connection string: %s", sanitize_connection_string(conn_str))
-
+                log('warning', f"Ignoring unknown connection parameter from kwargs: {key}")
+
+        # Step 4: Build connection string with merged params
+        builder = _ConnectionStringBuilder(normalized_params)
+
+        # Step 5: Add Driver and APP parameters (always controlled by the driver)
+        # These maintain existing behavior: Driver is always hardcoded, APP is always MSSQL-Python
+        builder.add_param('Driver', 'ODBC Driver 18 for SQL Server')
+        builder.add_param('APP', 'MSSQL-Python')
+
+        # Step 6: Build final string
+        conn_str = builder.build()
+
+        log('info', "Final connection string: %s", sanitize_connection_string(conn_str))
+
         return conn_str
 
     @property

mssql_python/connection_string_builder.py

@@ -0,0 +1,113 @@
+"""
+Copyright (c) Microsoft Corporation.
+Licensed under the MIT license.
+
+Connection string builder for mssql-python.
+
+Reconstructs ODBC connection strings from parameter dictionaries
+with proper escaping and formatting per MS-ODBCSTR specification.
+"""
+
+from typing import Dict, Optional
+from mssql_python.constants import _CONNECTION_STRING_DRIVER_KEY
+
+class _ConnectionStringBuilder:
+    """
+    Internal builder for ODBC connection strings. Not part of public API.
+
+    Handles proper escaping of special characters and reconstructs
+    connection strings in ODBC format.
+    """
+
+    def __init__(self, initial_params: Optional[Dict[str, str]] = None):
+        """
+        Initialize the builder with optional initial parameters.
+
+        Args:
+            initial_params: Dictionary of initial connection parameters
+        """
+        self._params: Dict[str, str] = initial_params.copy() if initial_params else {}
+
+    def add_param(self, key: str, value: str) -> '_ConnectionStringBuilder':
+        """
+        Add or update a connection parameter.
+
+        Args:
+            key: Parameter name (should be normalized canonical name)
+            value: Parameter value
+
+        Returns:
+            Self for method chaining
+        """
+        self._params[key] = str(value)
+        return self
+
+    def build(self) -> str:
+        """
+        Build the final connection string.
+
+        Returns:
+            ODBC-formatted connection string with proper escaping
+
+        Note:
+            - Driver parameter is placed first
+            - Other parameters are sorted for consistency
+            - Values are escaped if they contain special characters
+        """
+        parts = []
+
+        # Build in specific order: Driver first, then others
+        if _CONNECTION_STRING_DRIVER_KEY in self._params:
+            parts.append(f"Driver={self._escape_value(self._params['Driver'])}")
+
+        # Add other parameters (sorted for consistency)
+        for key in sorted(self._params.keys()):
+            if key == 'Driver':
+                continue  # Already added
+
+            value = self._params[key]
+            escaped_value = self._escape_value(value)
+            parts.append(f"{key}={escaped_value}")
+
+        # Join with semicolons
+        return ';'.join(parts)
+
+    def _escape_value(self, value: str) -> str:
+        """
+        Escape a parameter value if it contains special characters.
+
+        Per MS-ODBCSTR specification:
+        - Values containing ';', '{', '}', '=', or spaces should be braced for safety
+        - '}' inside braced values is escaped as '}}'
+        - '{' inside braced values is escaped as '{{'
+
+        Args:
+            value: Parameter value to escape
+
+        Returns:
+            Escaped value (possibly wrapped in braces)
+
+        Examples:
+            >>> builder = _ConnectionStringBuilder()
+            >>> builder._escape_value("localhost")
+            'localhost'
+            >>> builder._escape_value("local;host")
+            '{local;host}'
+            >>> builder._escape_value("p}w{d")
+            '{p}}w{{d}'
+            >>> builder._escape_value("ODBC Driver 18 for SQL Server")
+            '{ODBC Driver 18 for SQL Server}'
+        """
+        if not value:
+            return value
+
+        # Check if value contains special characters that require bracing
+        # Include spaces and = for safety, even though technically not always required
+        needs_braces = any(ch in value for ch in ';{}= ')
+
+        if needs_braces:
+            # Escape existing braces by doubling them
+            escaped = value.replace('}', '}}').replace('{', '{{')
+            return f'{{{escaped}}}'
+        else:
+            return value