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()

```

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)

{"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"}]}