jason810496 commented on code in PR #54597:
URL: https://github.com/apache/airflow/pull/54597#discussion_r2287283622
##########
airflow-core/src/airflow/api_fastapi/core_api/services/public/pools.py:
##########
@@ -103,25 +103,34 @@ def handle_bulk_update(self, action:
BulkUpdateAction[PoolBody], results: BulkAc
update_pool_names = to_update_pool_names
for pool in action.entities:
- if pool.pool in update_pool_names:
- old_pool =
self.session.scalar(select(Pool).filter(Pool.pool == pool.pool).limit(1))
+ if pool.pool not in update_pool_names:
+ continue
- data = {
- key: val for key, val in
pool.model_dump(by_alias=True).items() if val is not None
- }
- try:
- PoolBody(**data)
+ old_pool = self.session.scalar(select(Pool).filter(Pool.pool
== pool.pool).limit(1))
+ if not old_pool:
+ continue # Should not happen because we filtered above
- for key, val in data.items():
- setattr(old_pool, key, val)
+ # Dump Pydantic model data
+ data = pool.model_dump(by_alias=True)
+ fields_set = pool.model_fields_set
- results.success.append(str(pool.pool))
+ # Apply update_mask if provided
+ if action.update_mask:
+ fields_set = fields_set.intersection(action.update_mask)
- except ValidationError as e:
- results.errors.append({"error": f"{e.errors()}"})
+ # Update only fields in fields_set
+ for key in fields_set:
+ if key in data:
+ setattr(old_pool, key, data[key])
Review Comment:
It seems we can reuse the logic here:
https://github.com/apache/airflow/blob/6e92943a9651963a3af478c9f26265f703ad8a06/airflow-core/src/airflow/api_fastapi/core_api/routes/public/pools.py#L163-L175
How about having a common function to handle the patch utility for specified
Pool?
##########
airflow-core/src/airflow/api_fastapi/core_api/services/public/variables.py:
##########
@@ -94,18 +96,29 @@ def handle_bulk_update(self, action: BulkUpdateAction,
results: BulkActionRespon
update_keys = to_update_keys
for variable in action.entities:
- if variable.key in update_keys:
- old_variable =
self.session.scalar(select(Variable).filter_by(key=variable.key).limit(1))
+ if variable.key not in update_keys:
+ continue
+ old_variable =
self.session.scalar(select(Variable).filter_by(key=variable.key).limit(1))
+ if not old_variable:
+ continue # shouldn't happen because we filtered above
+
+ fields_to_update = variable.model_fields_set
+ if update_mask:
+ print(f"Update mask provided: {update_mask}")
+ # Only keep fields in update_mask
+ fields_to_update =
fields_to_update.intersection(update_mask)
+ else:
VariableBody(**variable.model_dump())
- data = variable.model_dump(exclude={"key"}, by_alias=True)
- for key, val in data.items():
- setattr(old_variable, key, val)
- results.success.append(variable.key)
+ data = variable.model_dump(include=fields_to_update -
non_update_fields, by_alias=True)
+ # Apply updates
Review Comment:
Same here, we could have common utility in service level for patching each
entity.
https://github.com/apache/airflow/blob/6e92943a9651963a3af478c9f26265f703ad8a06/airflow-core/src/airflow/api_fastapi/core_api/routes/public/variables.py#L144-L163
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/hitl.py:
##########
@@ -96,6 +96,7 @@ def _update_hitl_detail(
user: GetUserDep,
session: SessionDep,
map_index: int | None = None,
+ update_mask: list[str] | None = None, # <--- added update_mask
Review Comment:
```suggestion
update_mask: list[str] | None = Query(None),
```
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/xcom.py:
##########
@@ -314,10 +314,10 @@ def update_xcom_entry(
xcom_key: str,
patch_body: XComUpdateBody,
session: SessionDep,
+ update_mask: list[str] | None = Query(None),
) -> XComResponseNative:
- """Update an existing XCom entry."""
- # Check if XCom entry exists
- xcom_new_value = XComModel.serialize_value(patch_body.value)
+ """Update an existing XCom entry, partially if update_mask is provided."""
+ # Find the XCom entry
Review Comment:
Would it be better to not support `update_mask` but remain the exited
implementation for the `update_xcom_entry` route?
Since the `patch_body.map_index` is used for filter the `XComModel`, and
only only field to be update here is `patch_body.value`.
https://github.com/apache/airflow/blob/6e92943a9651963a3af478c9f26265f703ad8a06/airflow-core/src/airflow/api_fastapi/core_api/datamodels/xcom.py#L71-L76
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/hitl.py:
##########
@@ -196,6 +215,7 @@ def update_hitl_detail(
update_hitl_detail_payload: UpdateHITLDetailPayload,
user: GetUserDep,
session: SessionDep,
+ update_mask: list[str] | None = Query(None), # <--- added update_mask
Review Comment:
```suggestion
update_mask: list[str] | None = Query(None),
```
##########
airflow-core/src/airflow/api_fastapi/core_api/services/public/variables.py:
##########
@@ -81,6 +81,8 @@ def handle_bulk_update(self, action: BulkUpdateAction,
results: BulkActionRespon
"""Bulk Update variables."""
to_update_keys = {variable.key for variable in action.entities}
matched_keys, not_found_keys = self.categorize_keys(to_update_keys)
+ update_mask = getattr(action, "update_mask", None)
Review Comment:
```suggestion
update_mask = action.update_mask
```
--
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]
