Fix XCom get_value outside task runner - #68789

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide
Here are some useful points:

• Pay attention to the quality of your code (ruff, mypy and type annotations). Our prek-hooks will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example Dag that shows how users should use it.
• Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
• Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: [email protected]
  Slack: https://s.apache.org/airflow-slack

Hi, I don't think this is the right approach. The task-sdk should not be accessing the database directly. I would take a look at the docs here for a bit more detail on database access restrictions:
https://airflow.apache.org/docs/apache-airflow/stable/installation/upgrading_to_airflow3.html#database-access-restrictions

Updated the PR in cee3476 to avoid DB access from task-sdk. The revised patch restores BaseXCom to the supervisor-comms-only path, adds server-side XComModel.get_one()/get_value() helpers, exposes XComModel from the common compat SDK for plugin extra links, and adds /links route coverage for a plugin-registered extra link reading XCom through XComModel.

Updated in 5a6f988.

XComModel.get_one() / get_value() now keeps the DB lookup server-side in airflow-core, but routes deserialization through the configured custom XCom backend when one is configured. The default BaseXCom path still uses XComModel.deserialize_value() so existing JSON rows keep their current behavior.

I also added XComObjectStorageBackend regression coverage showing an object-storage-backed row returns the original value instead of the stored file path.

Verified locally on the rebased branch:

• uv run pytest --rootdir=providers/common/io providers/common/io/tests/unit/common/io/xcom/test_backend.py::TestXComObjectStorageBackend
• uv run pytest airflow-core/tests/unit/models/test_xcom.py airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_extra_links.py task-sdk/tests/task_sdk/bases/test_xcom.py
• uv run pytest --rootdir=providers/common/compat providers/common/compat/tests/unit/common/compat/test_sdk.py
• uv run ruff check airflow-core/src/airflow/models/xcom.py airflow-core/tests/unit/models/test_xcom.py airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_extra_links.py providers/common/compat/src/airflow/providers/common/compat/sdk.py providers/common/compat/tests/unit/common/compat/test_sdk.py providers/common/io/tests/unit/common/io/xcom/test_backend.py
• uv run ruff format --check airflow-core/src/airflow/models/xcom.py airflow-core/tests/unit/models/test_xcom.py airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_extra_links.py providers/common/compat/src/airflow/providers/common/compat/sdk.py providers/common/compat/tests/unit/common/compat/test_sdk.py providers/common/io/tests/unit/common/io/xcom/test_backend.py
• git diff --check

Updated the PR in cee3476 to avoid DB access from task-sdk. The revised patch restores BaseXCom to the supervisor-comms-only path, adds server-side XComModel.get_one()/get_value() helpers, exposes XComModel from the common compat SDK for plugin extra links, and adds /links route coverage for a plugin-registered extra link reading XCom through XComModel.

The new changes avoid the task-sdk db access issue, but I'm not sure exposing XComModel through the compat sdk is the correct approach either as it's mixing server and task imports in the same module. That being said, I'm not an expert here so I will defer to a maintainer.

Just curious, did you take a look at the proposed solution on the issue? It avoids the need for any of these changes by having the plugin use XComModel and create_session directly from core

Updated in 990456d to add the # use next version marker for common.io's common-compat dependency.

This addresses the Build info selective-check failure:

common.compat provider changed but the following providers don't have '# use next version'
  - common.io (providers/common/io/pyproject.toml)

Verified locally with:

uv run breeze ci selective-check --commit-ref HEAD --pr-labels "['area:task-sdk']" --github-event-name pull_request --github-repository apache/airflow --github-actor sdhilip200

The local check now reports:

common-compat-changed-without-next-version=false

Updated in 7534804 to make the common.compat sdk.XComModel test Airflow-version aware. The compat test now asserts XComModel on Airflow 3 and the XCom fallback on Airflow 2, matching the compat rename map. This addresses the Compat 2.11.1 provider distribution failure. Verified locally with: uv run pytest --rootdir=providers/common/compat providers/common/compat/tests/unit/common/compat/test_sdk.py; uv run ruff check providers/common/compat/tests/unit/common/compat/test_sdk.py; uv run ruff format --check providers/common/compat/tests/unit/common/compat/test_sdk.py; git diff --check.

Updated in 7ac3b14 to make the XComObjectStorageBackend regression run only when the installed Airflow exposes the new server-side XComModel helpers added by this PR. This keeps the object-storage backend regression active for the current implementation while preventing provider-compat jobs for released Airflow 3 wheels from failing on missing get_value or set(..., serialize=False) support. Verified locally with uv run pytest --rootdir=providers/common/io providers/common/io/tests/unit/common/io/xcom/test_backend.py -k xcommodel_get_value_uses_configured_backend. Also verified uv run ruff check providers/common/io/tests/unit/common/io/xcom/test_backend.py, uv run ruff format --check providers/common/io/tests/unit/common/io/xcom/test_backend.py, git diff --check, and uv run scripts/ci/prek/update_providers_dependencies.py with no generated-file diff.

Updated in d8b1907 to fix the static-check failure from check-airflow-v-imports-in-tests by importing AIRFLOW_V_3_0_PLUS from tests_common.test_utils.version_compat at module scope in the common.compat SDK test. Verified locally: python3 -m py_compile providers/common/compat/tests/unit/common/compat/test_sdk.py; uv run ruff check providers/common/compat/tests/unit/common/compat/test_sdk.py; uv run ruff format --check providers/common/compat/tests/unit/common/compat/test_sdk.py; uv run pytest --rootdir=providers/common/compat providers/common/compat/tests/unit/common/compat/test_sdk.py; git diff --check. I also inspected the provider-distribution compat failures in the same run; they fail while collecting providers/microsoft/azure/tests/unit/microsoft/azure/hooks/test_data_factory.py because FactoryListResponse is missing from azure.mgmt.datafactory.models, which is outside this PR's changed files.

This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions.

Does this need new API on XComModel? On #59093 the problem was using the Task SDK XCom in API-server code, and the existing XComModel.get_many(...) covers reading XComs there. get_one also calls deserialize_value on the result, which operatorlink.py deliberately avoids in the same API-server context because it "could instantiate arbitrary classes from untrusted XCom data". Exporting the server ORM model through common.compat.sdk mixes server and task imports in one module. Is there a case get_many doesn't cover? Converting to draft until there is.