justinpakzad commented on code in PR #59874:
URL: https://github.com/apache/airflow/pull/59874#discussion_r3724666114
##########
task-sdk/src/airflow/sdk/bases/xcom.py:
##########
@@ -572,3 +574,42 @@ def delete(
map_index=map_index,
),
)
+
+ @classmethod
+ def delete_all(
+ cls,
+ dag_id: str,
+ run_id: str,
+ task_id: str | None = None,
+ key: str | None = None,
+ map_index: int | None = None,
+ ) -> int:
+ """
+ Bulk delete XCom entries, optionally filtered by task_id, key, or
map_index.
+
+ :param dag_id: Dag ID.
+ :param run_id: Dag run ID for the task.
+ :param task_id: Optional task ID filter. If provided, only XComs from
this task
+ will be deleted. Pass *None* (default) to delete across all tasks.
+ :param key: Optional key filter. If provided, only XComs with this key
+ will be deleted. Pass *None* (default) to delete all keys.
+ :param map_index: Optional map index filter. If provided, only XComs
with this
+ map index will be deleted. Pass *None* (default) to delete all map
indexes.
+ :return: Number of XCom entries deleted.
+ """
+ from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
+
+ msg = SUPERVISOR_COMMS.send(
Review Comment:
I've updated the docstring and xcoms.rst:113 to indicate that this will not
purge any data for custom Xcom backends. Bulk purging can be implemented in a
follow up as I felt it would add another layer of complexity that would be
better suited for a separate PR.
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -392,6 +394,7 @@ def from_api_response(cls, response: HITLDetailResponse) ->
HITLDetailResponseRe
| GetVariableKeys
| PutVariable
| DeleteXCom
+ | BulkDeleteXCom
Review Comment:
Added it to the `ToTriggerRunner` and dropped it from the test exclusion.
##########
task-sdk/src/airflow/sdk/bases/xcom.py:
##########
@@ -572,3 +574,42 @@ def delete(
map_index=map_index,
),
)
+
+ @classmethod
+ def delete_all(
+ cls,
+ dag_id: str,
+ run_id: str,
+ task_id: str | None = None,
+ key: str | None = None,
+ map_index: int | None = None,
+ ) -> int:
+ """
+ Bulk delete XCom entries, optionally filtered by task_id, key, or
map_index.
+
+ :param dag_id: Dag ID.
+ :param run_id: Dag run ID for the task.
+ :param task_id: Optional task ID filter. If provided, only XComs from
this task
+ will be deleted. Pass *None* (default) to delete across all tasks.
+ :param key: Optional key filter. If provided, only XComs with this key
+ will be deleted. Pass *None* (default) to delete all keys.
+ :param map_index: Optional map index filter. If provided, only XComs
with this
+ map index will be deleted. Pass *None* (default) to delete all map
indexes.
+ :return: Number of XCom entries deleted.
Review Comment:
Good catch. I've added a new parameter `include_dag_result` which defaults
to `False`, skipping `dag_result` rows unless asked for.
--
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]
