pierrejeambrun commented on code in PR #54597:
URL: https://github.com/apache/airflow/pull/54597#discussion_r2316075424
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/hitl.py:
##########
@@ -64,7 +64,7 @@ def _get_task_instance(
dag_run_id: str,
task_id: str,
session: SessionDep,
- map_index: int,
+ map_index: int | None,
Review Comment:
This change does not seem related to this PR. Can you remove it please?
##########
airflow-core/src/airflow/api_fastapi/core_api/services/public/common.py:
##########
@@ -72,3 +72,56 @@ def handle_bulk_update(self, action: BulkUpdateAction[T],
results: BulkActionRes
def handle_bulk_delete(self, action: BulkDeleteAction[T], results:
BulkActionResponse) -> None:
"""Bulk delete entities."""
raise NotImplementedError
+
+
+class PatchUtil:
+ """
+ Utility class for applying patch operations with support for update masks.
+
+ This helper is used to update only selected fields of a Pydantic model
instance,
+ while ensuring:
+ - Only fields present in the update_mask are modified.
+ - Non-updatable fields can be excluded explicitly.
+ - Both raw field names and aliases are handled correctly.
+ """
+
+ @staticmethod
+ def apply_patch_with_update_mask(
+ model,
+ patch_body,
+ update_mask: list[str] | None,
+ non_update_fields: set[str] | None = None,
+ ):
+ """
+ Apply a patch to the given model using the provided update mask.
+
+ Args:
+ model: The model instance to update.
+ patch_body: Pydantic model containing patch data.
+ update_mask (list[str] | None): Optional list of fields to update.
+ non_update_fields (set[str] | None): Fields that should not be
updated.
+
+ Returns:
+ The updated model instance.
+
+ Raises:
+ HTTPException: If invalid fields are provided in update_mask.
Review Comment:
```suggestion
```
##########
airflow-core/src/airflow/api_fastapi/core_api/services/public/common.py:
##########
@@ -72,3 +72,56 @@ def handle_bulk_update(self, action: BulkUpdateAction[T],
results: BulkActionRes
def handle_bulk_delete(self, action: BulkDeleteAction[T], results:
BulkActionResponse) -> None:
"""Bulk delete entities."""
raise NotImplementedError
+
+
+class PatchUtil:
+ """
+ Utility class for applying patch operations with support for update masks.
+
+ This helper is used to update only selected fields of a Pydantic model
instance,
+ while ensuring:
+ - Only fields present in the update_mask are modified.
+ - Non-updatable fields can be excluded explicitly.
+ - Both raw field names and aliases are handled correctly.
+ """
+
+ @staticmethod
+ def apply_patch_with_update_mask(
+ model,
+ patch_body,
Review Comment:
missing type declaration.
##########
airflow-core/src/airflow/api_fastapi/core_api/services/public/variables.py:
##########
@@ -32,10 +34,61 @@
from airflow.api_fastapi.core_api.datamodels.variables import (
VariableBody,
)
-from airflow.api_fastapi.core_api.services.public.common import BulkService
+from airflow.api_fastapi.core_api.services.public.common import BulkService,
PatchUtil
from airflow.models.variable import Variable
+def update_orm_from_pydantic(
+ variable_key: str, patch_body: VariableBody, update_mask: list[str] |
None, session: SessionDep
+) -> Variable:
+ """
+ Patch an existing Variable with provided update fields.
+
+ Args:
+ variable_key (Variable_key):The name of the existing Variable_key to
update.
+ patch_body (VariableBody): The patch request body containing fields to
update.
+ update_mask (list[str] | None): List of fields to update. If None, all
provided fields will be updated.
+ session (SessionDep): The database session dependency.
+
+ Returns:
+ Variable: The updated Variable object.
+
+ Raises:
+ HTTPException: If attempting to update restricted fields (e.g., `key`).
Review Comment:
same everywhere.
##########
airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml:
##########
@@ -9076,6 +9096,16 @@ components:
type: array
title: Entities
description: A list of entities to be updated.
+ update_mask:
+ anyOf:
+ - items:
+ type: string
+ type: array
+ - type: 'null'
+ title: Update Mask
+ description: A list of field names to update for each entity.Only
these
+ fields will be applied from the request body to the database
model.Any
+ extra fields provided will be ignored.
Review Comment:
Formatting of the description is weird. There is no space after each dots.
##########
airflow-core/src/airflow/api_fastapi/core_api/services/public/pools.py:
##########
@@ -30,12 +32,84 @@
BulkUpdateAction,
)
from airflow.api_fastapi.core_api.datamodels.pools import (
+ BasePool,
PoolBody,
+ PoolPatchBody,
)
-from airflow.api_fastapi.core_api.services.public.common import BulkService
+from airflow.api_fastapi.core_api.services.public.common import BulkService,
PatchUtil
from airflow.models.pool import Pool
+def update_orm_from_pydantic(
+ pool_name: str,
+ patch_body: PoolBody | PoolPatchBody,
+ update_mask: list[str] | None,
+ session: SessionDep,
+) -> Pool:
+ """
+ Patch an existing Pool instance with provided update fields.
+
+ Args:
+ pool_name (str): The name of the existing Pool to be updated.
+ patch_body (PoolBody): Pydantic model containing the fields to update.
+ update_mask (list[str] | None): Specific fields to update. If None,
all provided fields will be considered.
+ session (SessionDep): The database session dependency.
+
+ Returns:
+ Pool: The updated Pool instance.
+
+ Raises:
+ HTTPException: If attempting to update disallowed fields on
`default_pool`.
Review Comment:
These docstrings are most likely generated and do not follow the appropriate
format. We don't use google docstring. You can either delete them or better
convert them to sphinx format. (no need to speficy types though, it's inferred
from type annotations)
##########
airflow-core/src/airflow/api_fastapi/core_api/services/public/common.py:
##########
@@ -72,3 +72,56 @@ def handle_bulk_update(self, action: BulkUpdateAction[T],
results: BulkActionRes
def handle_bulk_delete(self, action: BulkDeleteAction[T], results:
BulkActionResponse) -> None:
"""Bulk delete entities."""
raise NotImplementedError
+
+
+class PatchUtil:
+ """
+ Utility class for applying patch operations with support for update masks.
+
+ This helper is used to update only selected fields of a Pydantic model
instance,
+ while ensuring:
+ - Only fields present in the update_mask are modified.
+ - Non-updatable fields can be excluded explicitly.
+ - Both raw field names and aliases are handled correctly.
+ """
+
+ @staticmethod
+ def apply_patch_with_update_mask(
+ model,
+ patch_body,
+ update_mask: list[str] | None,
+ non_update_fields: set[str] | None = None,
+ ):
Review Comment:
missing return type declaration
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/hitl.py:
##########
@@ -95,7 +95,8 @@ def _update_hitl_detail(
update_hitl_detail_payload: UpdateHITLDetailPayload,
user: GetUserDep,
session: SessionDep,
- map_index: int,
+ map_index: int | None = None,
+ update_mask: list[str] | None = Query(None),
Review Comment:
```suggestion
update_mask: list[str] | None = Query(None),
map_index: int,
```
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/hitl.py:
##########
@@ -147,7 +166,7 @@ def _get_hitl_detail(
dag_run_id: str,
task_id: str,
session: SessionDep,
- map_index: int,
+ map_index: int | None = None,
Review Comment:
Why?
##########
airflow-core/src/airflow/api_fastapi/core_api/services/public/common.py:
##########
@@ -72,3 +72,56 @@ def handle_bulk_update(self, action: BulkUpdateAction[T],
results: BulkActionRes
def handle_bulk_delete(self, action: BulkDeleteAction[T], results:
BulkActionResponse) -> None:
"""Bulk delete entities."""
raise NotImplementedError
+
+
+class PatchUtil:
Review Comment:
I don't think we need a `PatchUtil` wrapper class at this point. a simple
`apply_patch_with_update_mask` module function should be enough, we can delete
this class.
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_pools.py:
##########
@@ -614,6 +635,29 @@ class TestBulkPools(TestPoolsEndpoint):
},
id="test_update_not_found",
),
+ pytest.param(
+ {
+ "actions": [
+ {
+ "action": "update",
+ "entities": [
+ {
+ "name": "pool1",
+ "slots": 50,
+ "description": "Updated description",
+ "include_deferred": False,
+ }
+ ],
+ "update_mask": ["slots", "description"],
+ "action_on_non_existence": "fail",
+ }
+ ]
+ },
+ {
+ "update": {"success": ["pool1"], "errors": []},
+ },
Review Comment:
I think we also need separate test to ensure that other attribute were not
modified, accordingly to the `update_mask` provided. (same for other entities)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
