Fix bulk create+overwrite silently resetting unset fields on pools and connections (#68645)

A bulk 'create' with action_on_existence=overwrite dumped the whole request body
and set every field on the existing record, so fields the request omitted were
reset to their defaults. For multi-team setups this silently nulled an existing
pool's or connection's team_name ownership when an overwrite changed only e.g.
slots; description and include_deferred were affected too.

Only write the fields the request actually provided (model_dump(exclude_unset=True)),
so omitted fields keep their current value while explicitly-set fields (even None)
are still applied. Adds regression tests for both pools and connections.

Author: seanghaeli

airflow-core/src/airflow/api_fastapi/core_api/services/public/connections.py

@@ -121,7 +121,12 @@ def handle_bulk_create(
                 if connection.connection_id in create_connection_ids:
                     if connection.connection_id in matched_connection_ids:
                         existed_connection = existed_connections_dict[connection.connection_id]
-                        for key, val in connection.model_dump(by_alias=True).items():
+                        # Only overwrite fields the request actually provided (see pools.py for the
+                        # full rationale). Plain ``model_dump()`` resets omitted fields to their
+                        # defaults on the existing connection — e.g. silently nulling ``team_name``
+                        # multi-team ownership. ``exclude_unset=True`` writes only the fields present
+                        # in the request body.
+                        for key, val in connection.model_dump(by_alias=True, exclude_unset=True).items():
                             setattr(existed_connection, key, val)
                     else:
                         self.session.add(Connection(**connection.model_dump(by_alias=True)))

airflow-core/src/airflow/api_fastapi/core_api/services/public/pools.py

@@ -149,7 +149,14 @@ def handle_bulk_create(self, action: BulkCreateAction[PoolBody], results: BulkAc
                 if pool.pool in create_pool_names:
                     if pool.pool in matched_pool_names:
                         existed_pool = existing_pools_dict[pool.pool]
-                        for key, val in pool.model_dump().items():
+                        # Only overwrite fields the request actually provided. Plain ``model_dump()``
+                        # emits every field at its default, so an overwrite that omits e.g.
+                        # ``team_name``/``description``/``include_deferred`` silently resets them on the
+                        # existing pool — most damagingly nulling its multi-team ``team_name`` ownership.
+                        # ``exclude_unset=True`` writes only fields present in the request body, so
+                        # omitted fields keep their current value while an explicitly-set field (even
+                        # ``None``) is still applied.
+                        for key, val in pool.model_dump(exclude_unset=True).items():
                             setattr(existed_pool, key, val)
                     else:
                         self.session.add(Pool(**pool.model_dump()))

airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_connections.py

@@ -1984,6 +1984,40 @@ def test_bulk_rejects_team_name_when_multi_team_is_disabled(self, test_client):
         expected_error_conn_ids = {err["input"]["connection_id"] for err in detail}
         assert sorted(expected_error_conn_ids) == ["test_conn_id_2", "test_conn_id_3"]
 
+    @conf_vars({("core", "multi_team"): "True"})
+    def test_bulk_create_overwrite_preserves_unset_team_name(self, test_client, testing_team, session):
+        """A bulk create+overwrite that omits ``team_name`` must NOT reset an existing connection's
+        ``team_name`` to ``None`` (parity with the pools fix). Overwriting with only ``conn_type``
+        previously clobbered every unset field via ``model_dump(by_alias=True)`` — silently nulling
+        the connection's multi-team ownership. ``exclude_unset=True`` preserves omitted fields.
+        """
+        self.create_connection(team_name=testing_team.name)
+        before = session.scalar(select(Connection).where(Connection.conn_id == TEST_CONN_ID))
+        assert before.team_name == testing_team.name
+
+        response = test_client.patch(
+            "/connections",
+            json={
+                "actions": [
+                    {
+                        "action": "create",
+                        "action_on_existence": "overwrite",
+                        "entities": [{"connection_id": TEST_CONN_ID, "conn_type": "new_type"}],
+                    }
+                ]
+            },
+        )
+        assert response.status_code == 200
+        assert response.json()["create"]["success"] == [TEST_CONN_ID]
+
+        session.expire_all()
+        after = session.scalar(select(Connection).where(Connection.conn_id == TEST_CONN_ID))
+        assert after.conn_type == "new_type"  # provided field is applied
+        assert after.team_name == testing_team.name, (
+            "bulk overwrite that omitted team_name must preserve existing ownership, "
+            f"got team_name={after.team_name!r}"
+        )
+
 
 class TestPostConnectionExtraBackwardCompatibility(TestConnectionEndpoint):
     def test_post_should_accept_empty_string_as_extra(self, test_client, session):

airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_pools.py

@@ -1240,3 +1240,68 @@ def test_bulk_rejects_team_name_when_multi_team_is_disabled(self, test_client):
 
         expected_error_names = {err["input"]["name"] for err in detail}
         assert sorted(expected_error_names) == ["pool_2", "pool_3"]
+
+    @conf_vars({("core", "multi_team"): "True"})
+    def test_bulk_create_overwrite_preserves_unset_team_name(self, test_client, session):
+        """A bulk create+overwrite that omits ``team_name`` must NOT reset an existing pool's
+        ``team_name`` to ``None``.
+
+        ``POOL1_NAME`` is owned by team ``test``. Overwriting it with a body that changes only
+        ``slots`` (no ``team_name``) previously clobbered every unset field back to its default via
+        ``model_dump()`` — silently nulling the pool's multi-team ownership. With
+        ``model_dump(exclude_unset=True)`` the omitted ``team_name`` keeps its current value.
+        """
+        self.create_pools()
+        before = session.scalar(select(Pool).where(Pool.pool == POOL1_NAME))
+        assert before.team_name == "test"
+
+        response = test_client.patch(
+            "/pools",
+            json={
+                "actions": [
+                    {
+                        "action": "create",
+                        "action_on_existence": "overwrite",
+                        "entities": [{"name": POOL1_NAME, "slots": 99}],
+                    }
+                ]
+            },
+        )
+        assert response.status_code == 200
+        assert response.json()["create"]["success"] == [POOL1_NAME]
+
+        session.expire_all()
+        after = session.scalar(select(Pool).where(Pool.pool == POOL1_NAME))
+        assert after.slots == 99  # the field that WAS provided is applied
+        assert after.team_name == "test", (
+            "bulk overwrite that omitted team_name must preserve existing ownership, "
+            f"got team_name={after.team_name!r}"
+        )
+
+    @conf_vars({("core", "multi_team"): "True"})
+    def test_bulk_create_overwrite_applies_explicit_team_name(self, test_client, session):
+        """An explicitly-provided ``team_name`` on a bulk overwrite is still applied (the fix only
+        skips *omitted* fields, it must not skip fields the request actually set)."""
+        _create_team()
+        session.add(Team(name="other"))
+        session.commit()
+        _create_pools()  # POOL1 owned by team "test"
+
+        response = test_client.patch(
+            "/pools",
+            json={
+                "actions": [
+                    {
+                        "action": "create",
+                        "action_on_existence": "overwrite",
+                        "entities": [{"name": POOL1_NAME, "slots": 7, "team_name": "other"}],
+                    }
+                ]
+            },
+        )
+        assert response.status_code == 200
+
+        session.expire_all()
+        after = session.scalar(select(Pool).where(Pool.pool == POOL1_NAME))
+        assert after.team_name == "other"
+        assert after.slots == 7