Fix schema alias handling in ConnectionResult (apache#46957)

Author: amoghrajesh

airflow/dag_processing/processor.py

@@ -277,7 +277,7 @@ def _handle_request(self, msg: ToManager, log: FilteringBoundLogger) -> None:  #
             conn = self.client.connections.get(msg.conn_id)
             if isinstance(conn, ConnectionResponse):
                 conn_result = ConnectionResult.from_conn_response(conn)
-                resp = conn_result.model_dump_json(exclude_unset=True).encode()
+                resp = conn_result.model_dump_json(exclude_unset=True, by_alias=True).encode()
             else:
                 resp = conn.model_dump_json().encode()
         elif isinstance(msg, GetVariable):

task_sdk/src/airflow/sdk/execution_time/comms.py

@@ -139,7 +139,9 @@ def from_conn_response(cls, connection_response: ConnectionResponse) -> Connecti
         # Exclude defaults to avoid sending unnecessary data
         # Pass the type as ConnectionResult explicitly so we can then call model_dump_json with exclude_unset=True
         # to avoid sending unset fields (which are defaults in our case).
-        return cls(**connection_response.model_dump(exclude_defaults=True), type="ConnectionResult")
+        return cls(
+            **connection_response.model_dump(exclude_defaults=True, by_alias=True), type="ConnectionResult"
+        )
 
 
 class VariableResult(VariableResponse):

task_sdk/src/airflow/sdk/execution_time/supervisor.py

@@ -787,7 +787,7 @@ def _handle_request(self, msg: ToSupervisor, log: FilteringBoundLogger):
             conn = self.client.connections.get(msg.conn_id)
             if isinstance(conn, ConnectionResponse):
                 conn_result = ConnectionResult.from_conn_response(conn)
-                resp = conn_result.model_dump_json(exclude_unset=True).encode()
+                resp = conn_result.model_dump_json(exclude_unset=True, by_alias=True).encode()
             else:
                 resp = conn.model_dump_json().encode()
         elif isinstance(msg, GetVariable):

task_sdk/tests/execution_time/test_supervisor.py

@@ -930,6 +930,15 @@ def watched_subprocess(self, mocker):
                 ConnectionResult(conn_id="test_conn", conn_type="mysql"),
                 id="get_connection",
             ),
+            pytest.param(
+                GetConnection(conn_id="test_conn"),
+                b'{"conn_id":"test_conn","conn_type":"mysql","schema":"mysql","type":"ConnectionResult"}\n',
+                "connections.get",
+                ("test_conn",),
+                {},
+                ConnectionResult(conn_id="test_conn", conn_type="mysql", schema="mysql"),  # type: ignore[call-arg]
+                id="get_connection_with_alias",
+            ),
             pytest.param(
                 GetVariable(key="test_key"),
                 b'{"key":"test_key","value":"test_value","type":"VariableResult"}\n',