Support connection extra parameters in MsSqlHook

providers/src/airflow/providers/microsoft/mssql/hooks/mssql.py

@@ -137,12 +137,14 @@ def _generate_insert_sql(self, table, values, target_fields, replace, **kwargs)
     def get_conn(self) -> PymssqlConnection:
         """Return ``pymssql`` connection object."""
         conn = self.connection
+        extra_conn_args = {key: val for key, val in conn.extra_dejson.items() if key != "sqlalchemy_scheme"}
         return pymssql.connect(
             server=conn.host,
             user=conn.login,
             password=conn.password,
             database=self.schema or conn.schema,
             port=str(conn.port),
+            **extra_conn_args,
         )
 
     def set_autocommit(

providers/tests/microsoft/mssql/hooks/test_mssql.py

@@ -55,6 +55,15 @@
     port=8081,
     extra={"SQlalchemy_Scheme": "mssql+testdriver", "myparam": "5@-//*"},
 )
+PYMSSQL_CONN_WITH_EXTRA = Connection(
+    conn_type="mssql",
+    host="test-server",
+    schema="test-db",
+    login="test-user",
+    password="test-password",
+    port=8081,
+    extra={"login_timeout": 30, "charset": "utf8", "tds_version": "7.0", "appname": "airflow"},
+)
 
 
 def get_primary_keys(self, table: str) -> list[str]:
@@ -79,6 +88,28 @@ def test_get_conn_should_return_connection(self, get_connection, mssql_get_conn)
         assert mssql_get_conn.return_value == conn
         mssql_get_conn.assert_called_once()
 
+    @mock.patch("airflow.providers.microsoft.mssql.hooks.mssql.pymssql.connect")
+    @mock.patch("airflow.providers.common.sql.hooks.sql.DbApiHook.get_connection")
+    def test_get_conn_with_extra_parameters(self, mock_get_connection, mock_connect):
+        mock_get_connection.return_value = PYMSSQL_CONN_WITH_EXTRA
+
+        hook = MsSqlHook()
+        hook.get_conn()
+
+        mock_connect.assert_called_once_with(
+            server="test-server",
+            user="test-user",
+            password="test-password",
+            database="test-db",
+            port="8081",
+            login_timeout=30,
+            charset="utf8",
+            tds_version="7.0",
+            appname="airflow",
+        )
+
+        assert hook.sqlalchemy_scheme == hook.DEFAULT_SQLALCHEMY_SCHEME
+
     @mock.patch("airflow.providers.microsoft.mssql.hooks.mssql.MsSqlHook.get_conn")
     @mock.patch("airflow.providers.common.sql.hooks.sql.DbApiHook.get_connection")
     def test_set_autocommit_should_invoke_autocommit(self, get_connection, mssql_get_conn):