Read only endpoint for XCom #8134

airflow/api_connexion/endpoints/xcom_endpoint.py

@@ -14,9 +14,18 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
+from flask import request
+from sqlalchemy import and_, func
+from sqlalchemy.orm.session import Session
 
-# TODO(mik-laj): We have to implement it.
-#     Do you want to help? Please look at: sshttps://github.com/apache/airflow/issues/8134
+from airflow.api_connexion import parameters
+from airflow.api_connexion.exceptions import NotFound
+from airflow.api_connexion.schemas.xcom_schema import (
+    XComCollection, XComCollectionItemSchema, XComCollectionSchema, xcom_collection_item_schema,
+    xcom_collection_schema,
+)
+from airflow.models import DagRun as DR, XCom
+from airflow.utils.session import provide_session
 
 
 def delete_xcom_entry():
@@ -26,18 +35,58 @@ def delete_xcom_entry():
     raise NotImplementedError("Not implemented yet.")
 
 
-def get_xcom_entries():
+@provide_session
+def get_xcom_entries(
+    dag_id: str,
+    dag_run_id: str,
+    task_id: str,
+    session: Session
+) -> XComCollectionSchema:
     """
     Get all XCom values
     """
-    raise NotImplementedError("Not implemented yet.")
+    offset = request.args.get(parameters.page_offset, 0)
+    limit = min(int(request.args.get(parameters.page_limit, 100)), 100)
+    query = session.query(XCom)
+    if dag_id != '~':
+        query = query.filter(XCom.dag_id == dag_id)
+        query.join(DR, and_(XCom.dag_id == DR.dag_id, XCom.execution_date == DR.execution_date))
+    else:
+        query.join(DR, XCom.execution_date == DR.execution_date)
+    if task_id != '~':
+        query = query.filter(XCom.task_id == task_id)
+    if dag_run_id != '~':
+        query = query.filter(DR.run_id == dag_run_id)
+    query = query.order_by(
+        XCom.execution_date, XCom.task_id, XCom.dag_id, XCom.key
+    )
+    total_entries = session.query(func.count(XCom.key)).scalar()
+    query = query.offset(offset).limit(limit)
+    return xcom_collection_schema.dump(XComCollection(xcom_entries=query.all(), total_entries=total_entries))
 
 
-def get_xcom_entry():
+@provide_session
+def get_xcom_entry(
+    dag_id: str,
+    task_id: str,
+    dag_run_id: str,
+    xcom_key: str,
+    session: Session
+) -> XComCollectionItemSchema:
     """
     Get an XCom entry
     """
-    raise NotImplementedError("Not implemented yet.")
+    query = session.query(XCom)
+    query = query.filter(and_(XCom.dag_id == dag_id,
+                              XCom.task_id == task_id,
+                              XCom.key == xcom_key))
+    query = query.join(DR, and_(XCom.dag_id == DR.dag_id, XCom.execution_date == DR.execution_date))
+    query = query.filter(DR.run_id == dag_run_id)
+
+    query_object = query.one_or_none()
+    if not query_object:
+        raise NotFound("XCom entry not found")
+    return xcom_collection_item_schema.dump(query_object)
 
 
 def patch_xcom_entry():

airflow/api_connexion/schemas/xcom_schema.py

@@ -0,0 +1,63 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from typing import List, NamedTuple
+
+from marshmallow import Schema, fields
+from marshmallow_sqlalchemy import SQLAlchemySchema, auto_field
+
+from airflow.models import XCom
+
+
+class XComCollectionItemSchema(SQLAlchemySchema):
+    """
+    Schema for a xcom item
+    """
+
+    class Meta:
+        """ Meta """
+        model = XCom
+
+    key = auto_field()
+    timestamp = auto_field()
+    execution_date = auto_field()
+    task_id = auto_field()
+    dag_id = auto_field()
+
+
+class XComSchema(XComCollectionItemSchema):
+    """
+    XCom schema
+    """
+
+    value = auto_field()
+
+
+class XComCollection(NamedTuple):
+    """ List of XComs with meta"""
+    xcom_entries: List[XCom]
+    total_entries: int
+
+
+class XComCollectionSchema(Schema):
+    """ XCom Collection Schema"""
+    xcom_entries = fields.List(fields.Nested(XComCollectionItemSchema))
+    total_entries = fields.Int()
+
+
+xcom_schema = XComSchema(strict=True)
+xcom_collection_item_schema = XComCollectionItemSchema(strict=True)
+xcom_collection_schema = XComCollectionSchema(strict=True)