Fix version_compat for Airflow 3.0.X

[morgan-dgk]

The decorated operator for this provide fails with Airflow 3.0.X and provider version > 4.4.0. #52465 updated version_compat and the operator to use the BaseOperator from the Task SDK. However, this operator is missing the required xcom_push method in version 3.0.X.

This causes failures when using the decorater operator with AttributeError: '_DockerDecoratedOperator' object has no attribute 'xcom_push'.

Resolves #53077

Tested locally with DAG below:

from airflow.sdk import dag, task

@dag(
   schedule=None,
   start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
   catchup=False,
   tags=["example"],
)
def docker_test():

   @task.docker(image="python:3.10-slim")
   def hello_world():
       return "Hello, world!"

   hello_world()


docker_test()

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

[morgan-dgk]

Worth noting that it looks like this could show up again.

The standardised version_compat provided in #52378 is the same code that was used in #52465. So this may not have been fully accounted for. However, I don't have full context / background on this.

[ashb]

Doing this based on the Airflow version isn't the right pattern really.

The reason is that the version that is important is actually the TaskSDK version, not the Airflow version, and with 3.1 we are working very hard towards making it possible to use operators etc without the main "apache-airflow-core" dist installed.

[uranusjr]

The PR should add a test that would fail in main without the change.

[morgan-dgk]

Doing this based on the Airflow version isn't the right pattern really.

The reason is that the version that is important is actually the TaskSDK version, not the Airflow version, and with 3.1 we are working very hard towards making it possible to use operators etc without the main "apache-airflow-core" dist installed.

OK, is there an existing example of the right way of doing this?

This code is reproduced unmodified from the Google Cloud provider version_compat file since I happened to see mention of the same issue in the comments there.

[ashb]

Yeah, we aren't very good about it yet.

In this case it's a bit hard I suspect, but this is probably a:

try:
    from airflow.sdk import Someting
except ImportError:
   from airflow.old import Else

but I don't know if that pattern will work here.

[morgan-dgk]

Yeah, we aren't very good about it yet.

In this case it's a bit hard I suspect, but this is probably a:

try:
    from airflow.sdk import Someting
except ImportError:
   from airflow.old import Else

but I don't know if that pattern will work here.

Here the BaseOperator class is already present in the TaskSDK, but is missing the xcom_push method from the BaseOperator from the old models import.

So, I don't think the above would work here.

[kaxil]

Worth noting that it looks like this could show up again.

The standardised version_compat provided in #52378 is the same code that was used in #52465. So this may not have been fully accounted for. However, I don't have full context / background on this.

Check #52378 (comment) . It include the logic of why 3.0 vs 3.1 specifically for DecoratedOperators

[kaxil]

Ideally we should have test for this but since this change is related to the DecoratedOperator in task SDK, I am merging it to unblock and have it released.