prompt: We are currently solving the following issue within our repository. Here is the issue text:
  --- BEGIN ISSUE ---
  API Endpoints - /xcomEntries/{xcom_key} cannot deserialize customized xcom backend
  ### Description
  
  We use S3 as our xcom backend database and write serialize/deserialize method for xcoms.
  However, when we want to access xcom through REST API, it returns the s3 file url instead of the deserialized value. Could you please add the feature to support customized xcom backend for REST API access?
  
  ### Use case/motivation
  
  _No response_
  
  ### Related issues
  
  _No response_
  
  ### Are you willing to submit a PR?
  
  - [ ] Yes I am willing to submit a PR!
  
  ### Code of Conduct
  
  - [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
  
  
  --- END ISSUE ---
  
  Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
  --- BEGIN FILES ---
  Path: `airflow/api_connexion/endpoints/xcom_endpoint.py`
  Content:
  ```
  1 # Licensed to the Apache Software Foundation (ASF) under one
  2 # or more contributor license agreements.  See the NOTICE file
  3 # distributed with this work for additional information
  4 # regarding copyright ownership.  The ASF licenses this file
  5 # to you under the Apache License, Version 2.0 (the
  6 # "License"); you may not use this file except in compliance
  7 # with the License.  You may obtain a copy of the License at
  8 #
  9 #   http://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 # Unless required by applicable law or agreed to in writing,
  12 # software distributed under the License is distributed on an
  13 # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  14 # KIND, either express or implied.  See the License for the
  15 # specific language governing permissions and limitations
  16 # under the License.
  17 from typing import Optional
  18 
  19 from flask import g
  20 from sqlalchemy import and_
  21 from sqlalchemy.orm import Session
  22 
  23 from airflow.api_connexion import security
  24 from airflow.api_connexion.exceptions import NotFound
  25 from airflow.api_connexion.parameters import check_limit, format_parameters
  26 from airflow.api_connexion.schemas.xcom_schema import XComCollection, xcom_collection_schema, xcom_schema
  27 from airflow.api_connexion.types import APIResponse
  28 from airflow.models import DagRun as DR, XCom
  29 from airflow.security import permissions
  30 from airflow.utils.airflow_flask_app import get_airflow_app
  31 from airflow.utils.session import NEW_SESSION, provide_session
  32 
  33 
  34 @security.requires_access(
  35     [
  36         (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG),
  37         (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG_RUN),
  38         (permissions.ACTION_CAN_READ, permissions.RESOURCE_TASK_INSTANCE),
  39         (permissions.ACTION_CAN_READ, permissions.RESOURCE_XCOM),
  40     ],
  41 )
  42 @format_parameters({"limit": check_limit})
  43 @provide_session
  44 def get_xcom_entries(
  45     *,
  46     dag_id: str,
  47     dag_run_id: str,
  48     task_id: str,
  49     limit: Optional[int],
  50     offset: Optional[int] = None,
  51     session: Session = NEW_SESSION,
  52 ) -> APIResponse:
  53     """Get all XCom values"""
  54     query = session.query(XCom)
  55     if dag_id == '~':
  56         appbuilder = get_airflow_app().appbuilder
  57         readable_dag_ids = appbuilder.sm.get_readable_dag_ids(g.user)
  58         query = query.filter(XCom.dag_id.in_(readable_dag_ids))
  59         query = query.join(DR, and_(XCom.dag_id == DR.dag_id, XCom.run_id == DR.run_id))
  60     else:
  61         query = query.filter(XCom.dag_id == dag_id)
  62         query = query.join(DR, and_(XCom.dag_id == DR.dag_id, XCom.run_id == DR.run_id))
  63 
  64     if task_id != '~':
  65         query = query.filter(XCom.task_id == task_id)
  66     if dag_run_id != '~':
  67         query = query.filter(DR.run_id == dag_run_id)
  68     query = query.order_by(DR.execution_date, XCom.task_id, XCom.dag_id, XCom.key)
  69     total_entries = query.count()
  70     query = query.offset(offset).limit(limit)
  71     return xcom_collection_schema.dump(XComCollection(xcom_entries=query.all(), total_entries=total_entries))
  72 
  73 
  74 @security.requires_access(
  75     [
  76         (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG),
  77         (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG_RUN),
  78         (permissions.ACTION_CAN_READ, permissions.RESOURCE_TASK_INSTANCE),
  79         (permissions.ACTION_CAN_READ, permissions.RESOURCE_XCOM),
  80     ],
  81 )
  82 @provide_session
  83 def get_xcom_entry(
  84     *,
  85     dag_id: str,
  86     task_id: str,
  87     dag_run_id: str,
  88     xcom_key: str,
  89     session: Session = NEW_SESSION,
  90 ) -> APIResponse:
  91     """Get an XCom entry"""
  92     query = session.query(XCom).filter(XCom.dag_id == dag_id, XCom.task_id == task_id, XCom.key == xcom_key)
  93     query = query.join(DR, and_(XCom.dag_id == DR.dag_id, XCom.run_id == DR.run_id))
  94     query = query.filter(DR.run_id == dag_run_id)
  95 
  96     query_object = query.one_or_none()
  97     if not query_object:
  98         raise NotFound("XCom entry not found")
  99     return xcom_schema.dump(query_object)
  100
  ```
  --- END FILES ---
  
  Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.
  
  Here is an example:
  
  ```diff
  diff --git a/examples/server_async.py b/examples/server_async.py
  --- a/examples/server_async.py
  +++ b/examples/server_async.py
  @@ -313,4 +313,4 @@
   
   
   if __name__ == "__main__":
  -    asyncio.run(run_async_server("."), debug=True)
  +    asyncio.run(run_async_server(), debug=True)
  diff --git a/examples/server_sync.py b/examples/server_sync.py
  --- a/examples/server_sync.py
  +++ b/examples/server_sync.py
  @@ -313,5 +313,5 @@
   
   
   if __name__ == "__main__":
  -    server = run_sync_server(".")
  +    server = run_sync_server()
       server.shutdown()
  
  ```
  
golden_diff: diff --git a/airflow/api_connexion/endpoints/xcom_endpoint.py b/airflow/api_connexion/endpoints/xcom_endpoint.py
  --- a/airflow/api_connexion/endpoints/xcom_endpoint.py
  +++ b/airflow/api_connexion/endpoints/xcom_endpoint.py
  @@ -14,6 +14,7 @@
   # KIND, either express or implied.  See the License for the
   # specific language governing permissions and limitations
   # under the License.
  +import copy
   from typing import Optional
   
   from flask import g
  @@ -68,7 +69,7 @@
       query = query.order_by(DR.execution_date, XCom.task_id, XCom.dag_id, XCom.key)
       total_entries = query.count()
       query = query.offset(offset).limit(limit)
  -    return xcom_collection_schema.dump(XComCollection(xcom_entries=query.all(), total_entries=total_entries))
  +    return xcom_collection_schema.dump(XComCollection(xcom_entries=query, total_entries=total_entries))
   
   
   @security.requires_access(
  @@ -86,14 +87,28 @@
       task_id: str,
       dag_run_id: str,
       xcom_key: str,
  +    deserialize: bool = False,
       session: Session = NEW_SESSION,
   ) -> APIResponse:
       """Get an XCom entry"""
  -    query = session.query(XCom).filter(XCom.dag_id == dag_id, XCom.task_id == task_id, XCom.key == xcom_key)
  +    if deserialize:
  +        query = session.query(XCom, XCom.value)
  +    else:
  +        query = session.query(XCom)
  +
  +    query = query.filter(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.run_id == DR.run_id))
       query = query.filter(DR.run_id == dag_run_id)
   
  -    query_object = query.one_or_none()
  -    if not query_object:
  +    item = query.one_or_none()
  +    if item is None:
           raise NotFound("XCom entry not found")
  -    return xcom_schema.dump(query_object)
  +
  +    if deserialize:
  +        xcom, value = item
  +        stub = copy.copy(xcom)
  +        stub.value = value
  +        stub.value = XCom.deserialize_value(stub)
  +        item = stub
  +
  +    return xcom_schema.dump(item)
  
verification_info: {"golden_diff": "diff --git a/airflow/api_connexion/endpoints/xcom_endpoint.py b/airflow/api_connexion/endpoints/xcom_endpoint.py\n--- a/airflow/api_connexion/endpoints/xcom_endpoint.py\n+++ b/airflow/api_connexion/endpoints/xcom_endpoint.py\n@@ -14,6 +14,7 @@\n # KIND, either express or implied.  See the License for the\n # specific language governing permissions and limitations\n # under the License.\n+import copy\n from typing import Optional\n \n from flask import g\n@@ -68,7 +69,7 @@\n     query = query.order_by(DR.execution_date, XCom.task_id, XCom.dag_id, XCom.key)\n     total_entries = query.count()\n     query = query.offset(offset).limit(limit)\n-    return xcom_collection_schema.dump(XComCollection(xcom_entries=query.all(), total_entries=total_entries))\n+    return xcom_collection_schema.dump(XComCollection(xcom_entries=query, total_entries=total_entries))\n \n \n @security.requires_access(\n@@ -86,14 +87,28 @@\n     task_id: str,\n     dag_run_id: str,\n     xcom_key: str,\n+    deserialize: bool = False,\n     session: Session = NEW_SESSION,\n ) -> APIResponse:\n     \"\"\"Get an XCom entry\"\"\"\n-    query = session.query(XCom).filter(XCom.dag_id == dag_id, XCom.task_id == task_id, XCom.key == xcom_key)\n+    if deserialize:\n+        query = session.query(XCom, XCom.value)\n+    else:\n+        query = session.query(XCom)\n+\n+    query = query.filter(XCom.dag_id == dag_id, XCom.task_id == task_id, XCom.key == xcom_key)\n     query = query.join(DR, and_(XCom.dag_id == DR.dag_id, XCom.run_id == DR.run_id))\n     query = query.filter(DR.run_id == dag_run_id)\n \n-    query_object = query.one_or_none()\n-    if not query_object:\n+    item = query.one_or_none()\n+    if item is None:\n         raise NotFound(\"XCom entry not found\")\n-    return xcom_schema.dump(query_object)\n+\n+    if deserialize:\n+        xcom, value = item\n+        stub = copy.copy(xcom)\n+        stub.value = value\n+        stub.value = XCom.deserialize_value(stub)\n+        item = stub\n+\n+    return xcom_schema.dump(item)\n", "issue": "API Endpoints - /xcomEntries/{xcom_key} cannot deserialize customized xcom backend\n### Description\n\nWe use S3 as our xcom backend database and write serialize/deserialize method for xcoms.\r\nHowever, when we want to access xcom through REST API, it returns the s3 file url instead of the deserialized value. Could you please add the feature to support customized xcom backend for REST API access?\n\n### Use case/motivation\n\n_No response_\n\n### Related issues\n\n_No response_\n\n### Are you willing to submit a PR?\n\n- [ ] Yes I am willing to submit a PR!\n\n### Code of Conduct\n\n- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)\n\n", "before_files": [{"content": "# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  You may obtain a copy of the License at\n#\n#   http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\nfrom typing import Optional\n\nfrom flask import g\nfrom sqlalchemy import and_\nfrom sqlalchemy.orm import Session\n\nfrom airflow.api_connexion import security\nfrom airflow.api_connexion.exceptions import NotFound\nfrom airflow.api_connexion.parameters import check_limit, format_parameters\nfrom airflow.api_connexion.schemas.xcom_schema import XComCollection, xcom_collection_schema, xcom_schema\nfrom airflow.api_connexion.types import APIResponse\nfrom airflow.models import DagRun as DR, XCom\nfrom airflow.security import permissions\nfrom airflow.utils.airflow_flask_app import get_airflow_app\nfrom airflow.utils.session import NEW_SESSION, provide_session\n\n\n@security.requires_access(\n    [\n        (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG),\n        (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG_RUN),\n        (permissions.ACTION_CAN_READ, permissions.RESOURCE_TASK_INSTANCE),\n        (permissions.ACTION_CAN_READ, permissions.RESOURCE_XCOM),\n    ],\n)\n@format_parameters({\"limit\": check_limit})\n@provide_session\ndef get_xcom_entries(\n    *,\n    dag_id: str,\n    dag_run_id: str,\n    task_id: str,\n    limit: Optional[int],\n    offset: Optional[int] = None,\n    session: Session = NEW_SESSION,\n) -> APIResponse:\n    \"\"\"Get all XCom values\"\"\"\n    query = session.query(XCom)\n    if dag_id == '~':\n        appbuilder = get_airflow_app().appbuilder\n        readable_dag_ids = appbuilder.sm.get_readable_dag_ids(g.user)\n        query = query.filter(XCom.dag_id.in_(readable_dag_ids))\n        query = query.join(DR, and_(XCom.dag_id == DR.dag_id, XCom.run_id == DR.run_id))\n    else:\n        query = query.filter(XCom.dag_id == dag_id)\n        query = query.join(DR, and_(XCom.dag_id == DR.dag_id, XCom.run_id == DR.run_id))\n\n    if task_id != '~':\n        query = query.filter(XCom.task_id == task_id)\n    if dag_run_id != '~':\n        query = query.filter(DR.run_id == dag_run_id)\n    query = query.order_by(DR.execution_date, XCom.task_id, XCom.dag_id, XCom.key)\n    total_entries = query.count()\n    query = query.offset(offset).limit(limit)\n    return xcom_collection_schema.dump(XComCollection(xcom_entries=query.all(), total_entries=total_entries))\n\n\n@security.requires_access(\n    [\n        (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG),\n        (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG_RUN),\n        (permissions.ACTION_CAN_READ, permissions.RESOURCE_TASK_INSTANCE),\n        (permissions.ACTION_CAN_READ, permissions.RESOURCE_XCOM),\n    ],\n)\n@provide_session\ndef get_xcom_entry(\n    *,\n    dag_id: str,\n    task_id: str,\n    dag_run_id: str,\n    xcom_key: str,\n    session: Session = NEW_SESSION,\n) -> APIResponse:\n    \"\"\"Get an XCom entry\"\"\"\n    query = session.query(XCom).filter(XCom.dag_id == dag_id, XCom.task_id == task_id, XCom.key == xcom_key)\n    query = query.join(DR, and_(XCom.dag_id == DR.dag_id, XCom.run_id == DR.run_id))\n    query = query.filter(DR.run_id == dag_run_id)\n\n    query_object = query.one_or_none()\n    if not query_object:\n        raise NotFound(\"XCom entry not found\")\n    return xcom_schema.dump(query_object)\n", "path": "airflow/api_connexion/endpoints/xcom_endpoint.py"}], "after_files": [{"content": "# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  You may obtain a copy of the License at\n#\n#   http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\nimport copy\nfrom typing import Optional\n\nfrom flask import g\nfrom sqlalchemy import and_\nfrom sqlalchemy.orm import Session\n\nfrom airflow.api_connexion import security\nfrom airflow.api_connexion.exceptions import NotFound\nfrom airflow.api_connexion.parameters import check_limit, format_parameters\nfrom airflow.api_connexion.schemas.xcom_schema import XComCollection, xcom_collection_schema, xcom_schema\nfrom airflow.api_connexion.types import APIResponse\nfrom airflow.models import DagRun as DR, XCom\nfrom airflow.security import permissions\nfrom airflow.utils.airflow_flask_app import get_airflow_app\nfrom airflow.utils.session import NEW_SESSION, provide_session\n\n\n@security.requires_access(\n    [\n        (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG),\n        (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG_RUN),\n        (permissions.ACTION_CAN_READ, permissions.RESOURCE_TASK_INSTANCE),\n        (permissions.ACTION_CAN_READ, permissions.RESOURCE_XCOM),\n    ],\n)\n@format_parameters({\"limit\": check_limit})\n@provide_session\ndef get_xcom_entries(\n    *,\n    dag_id: str,\n    dag_run_id: str,\n    task_id: str,\n    limit: Optional[int],\n    offset: Optional[int] = None,\n    session: Session = NEW_SESSION,\n) -> APIResponse:\n    \"\"\"Get all XCom values\"\"\"\n    query = session.query(XCom)\n    if dag_id == '~':\n        appbuilder = get_airflow_app().appbuilder\n        readable_dag_ids = appbuilder.sm.get_readable_dag_ids(g.user)\n        query = query.filter(XCom.dag_id.in_(readable_dag_ids))\n        query = query.join(DR, and_(XCom.dag_id == DR.dag_id, XCom.run_id == DR.run_id))\n    else:\n        query = query.filter(XCom.dag_id == dag_id)\n        query = query.join(DR, and_(XCom.dag_id == DR.dag_id, XCom.run_id == DR.run_id))\n\n    if task_id != '~':\n        query = query.filter(XCom.task_id == task_id)\n    if dag_run_id != '~':\n        query = query.filter(DR.run_id == dag_run_id)\n    query = query.order_by(DR.execution_date, XCom.task_id, XCom.dag_id, XCom.key)\n    total_entries = query.count()\n    query = query.offset(offset).limit(limit)\n    return xcom_collection_schema.dump(XComCollection(xcom_entries=query, total_entries=total_entries))\n\n\n@security.requires_access(\n    [\n        (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG),\n        (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG_RUN),\n        (permissions.ACTION_CAN_READ, permissions.RESOURCE_TASK_INSTANCE),\n        (permissions.ACTION_CAN_READ, permissions.RESOURCE_XCOM),\n    ],\n)\n@provide_session\ndef get_xcom_entry(\n    *,\n    dag_id: str,\n    task_id: str,\n    dag_run_id: str,\n    xcom_key: str,\n    deserialize: bool = False,\n    session: Session = NEW_SESSION,\n) -> APIResponse:\n    \"\"\"Get an XCom entry\"\"\"\n    if deserialize:\n        query = session.query(XCom, XCom.value)\n    else:\n        query = session.query(XCom)\n\n    query = query.filter(XCom.dag_id == dag_id, XCom.task_id == task_id, XCom.key == xcom_key)\n    query = query.join(DR, and_(XCom.dag_id == DR.dag_id, XCom.run_id == DR.run_id))\n    query = query.filter(DR.run_id == dag_run_id)\n\n    item = query.one_or_none()\n    if item is None:\n        raise NotFound(\"XCom entry not found\")\n\n    if deserialize:\n        xcom, value = item\n        stub = copy.copy(xcom)\n        stub.value = value\n        stub.value = XCom.deserialize_value(stub)\n        item = stub\n\n    return xcom_schema.dump(item)\n", "path": "airflow/api_connexion/endpoints/xcom_endpoint.py"}]}