kaxil commented on a change in pull request #9482: URL: https://github.com/apache/airflow/pull/9482#discussion_r443906611
########## File path: airflow/api_connexion/endpoints/xcom_endpoint.py ########## @@ -14,16 +14,45 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. +from typing import List, Optional -# TODO(mik-laj): We have to implement it. -# Do you want to help? Please look at: sshttps://github.com/apache/airflow/issues/8134 +from flask import Response, request +from marshmallow import ValidationError +from sqlalchemy.orm.session import Session +from airflow.api_connexion.exceptions import AlreadyExists, BadRequest, NotFound +from airflow.api_connexion.schemas.xcom_schema import xcom_schema +from airflow.models import DagRun as DR, XCom +from airflow.utils.session import provide_session -def delete_xcom_entry(): + +@provide_session +def delete_xcom_entry( + dag_id: str, + dag_run_id: str, + task_id: str, + xcom_key: str, + session: Session +) -> Response: """ Delete an XCom entry """ - raise NotImplementedError("Not implemented yet.") + + dag_run = session.query(DR).filter(DR.run_id == dag_run_id, + DR.dag_id == dag_id).first() + if not dag_run: + raise NotFound(f'DAGRun with dag_id:{dag_id} and run_id:{dag_run_id} not found') + + query = session.query(XCom) + query = query.filter(XCom.dag_id == dag_id, + XCom.task_id == task_id, + XCom.key == xcom_key) + entry = query.delete() Review comment: You can chain these SQL Queries as I have done here: https://github.com/apache/airflow/pull/9424 ---------------------------------------------------------------- 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
