diff --git a/airflow-core/src/airflow/api_fastapi/core_api/services/public/connections.py b/airflow-core/src/airflow/api_fastapi/core_api/services/public/connections.py index bc6e5e49f75b7..74f5d05df776e 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/services/public/connections.py +++ b/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))) diff --git a/airflow-core/src/airflow/api_fastapi/core_api/services/public/pools.py b/airflow-core/src/airflow/api_fastapi/core_api/services/public/pools.py index dce54099772de..476c45b8f5ccf 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/services/public/pools.py +++ b/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())) diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_connections.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_connections.py index 6930039e1abbc..74920b11d51fd 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_connections.py +++ b/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): diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_pools.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_pools.py index 6e17598f07ebc..a383713fe8866 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_pools.py +++ b/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