Centralize version-aware EmptyOperator imports in a compatibility module

[pankajkoti]

Introduces cosmos/airflow/compatibility.py, a small module that centralizes the version-specific import of Airflow objects whose path changed between Airflow 2 and 3. The first such object is EmptyOperator, which moved to the standard provider in Airflow 3. The legacy airflow.operators.empty path still resolves on Airflow 3 but emits a DeprecatedImportWarning, so the module imports from the standard provider on Airflow 3 and from the legacy path on Airflow 2. The selection compares on the Airflow major version, so pre-releases such as 3.0.0rc1 are treated as Airflow 3.

The module also exposes EMPTY_OPERATOR_CLASS_PATH, the dotted import path for the version-appropriate EmptyOperator, derived from the imported class so it always stays in sync.

Call sites reference the operator by name or via the constant instead of duplicating the version check:

• watcher_kubernetes.py and _watcher/base.py import EmptyOperator from the module, replacing their duplicated try/except import blocks.
• graph.py serializes the operator_class string via EMPTY_OPERATOR_CLASS_PATH.

🤖 Generated with Claude Code

Co-authored-by: Tatiana Al-Chueyr tatiana.alchueyr@gmail.com

[Copilot]

Pull request overview

Centralizes the EmptyOperator import path into a version-aware EMPTY_OPERATOR_CLASS constant to avoid emitting DeprecatedImportWarning on Airflow 3.

Changes:

• Add EMPTY_OPERATOR_CLASS constant in cosmos/constants.py selecting the appropriate path based on Airflow version.
• Replace the hardcoded legacy path in cosmos/airflow/graph.py with the new constant.
• Update the corresponding test parametrization to use the constant.

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated no comments.

File	Description
cosmos/constants.py	Introduces version-aware EMPTY_OPERATOR_CLASS constant.
cosmos/airflow/graph.py	Uses the new constant in source-without-freshness rendering.
tests/airflow/test_graph.py	Updates expected operator class in parametrized test.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 98.28%. Comparing base (c8e1b69) to head (c0830b2).

Additional details and impacted files

@@           Coverage Diff           @@
##             main    #2758   +/-   ##
=======================================
  Coverage   98.28%   98.28%           
=======================================
  Files         106      107    +1     
  Lines        7911     7915    +4     
=======================================
+ Hits         7775     7779    +4     
  Misses        136      136           

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[tatiana]

Hi @pankajkoti ,

I feel uneasy about using a variable to represent a well-known class. This tends to make the human maintenance of the code harder.

That said, I agree we should unify how we handle different import paths (AF2/AF3) across our codebase.

Please, could you take a look at this library - it may be a better solution:
https://airflow.apache.org/docs/apache-airflow-providers-common-compat/stable/index.html

If it is not, I suggest we create a module in cosmos/airflow/compatibility.py to handle the import, keeping the class name as EmptyOperator.

[pankajkoti]

Thanks @tatiana for the quick review.

Please, could you take a look at this library - it may be a better solution:
https://airflow.apache.org/docs/apache-airflow-providers-common-compat/stable/index.html
If it is not, I suggest we create a module in cosmos/airflow/compatibility.py to handle the import, keeping the class name as EmptyOperator.

I checked apache-airflow-providers-common-compat, and it does not ship an EmptyOperator. Its standard/operators.py only re-exports PythonOperator, ShortCircuitOperator, and a few SDK base classes, so common-compat is not an option here.

I went with your second suggestion: a new cosmos/airflow/compatibility.py that handles the version-specific import and keeps the name as EmptyOperator. It resolves names lazily (PEP 562 module __getattr__), mirroring the pattern the standard provider itself uses, so importing the module stays cheap and call sites reference the real class rather than a string variable.

Changes in the latest push:

• watcher_kubernetes.py and _watcher/base.py now import EmptyOperator from the new module instead of duplicating try/except blocks.
• graph.py builds the serialized operator_class path via get_version_aware_operator_class_path(EmptyOperator).
• The EMPTY_OPERATOR_CLASS constant is removed.

The module is intentionally generic so we can fold other Airflow 2/3 import splits into it over time.

[Copilot]

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated 2 comments.

[tatiana]

@pankajkoti approving, so you can move forward with the other tasks. Even though we have different views on the approach, this should not block your progress.

[Copilot]

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated 1 comment.

[pankajkoti]

@pankajkoti approving, so you can move forward with the other tasks. Even though we have different views on the approach, this should not block your progress.

Thanks @tatiana. I've pushed the simpler version: a direct version-gated import in cosmos/airflow/compatibility.py plus an EMPTY_OPERATOR_CLASS_PATH constant derived from the imported class, and removed the lazy __getattr__ layer and the helper. graph.py now serialises the path via that constant.

I checked the full apache-airflow-providers-common-compat provider. The only operators it ships are PythonOperator and ShortCircuitOperator (alongside TimeDeltaTrigger, BaseNotifier, a few security permission constants, and OpenLineage helpers). It does not cover EmptyOperator, nor the BaseOperator and TaskGroup Airflow 2/3 splits, nor any of the heavy provider operators we guard (Docker, AWS ECS, GCP Cloud Run, Azure Container Instances, BigQuery, KubernetesPod). So common-compat would not absorb the imports we keep duplicating, and we would still need our own shim.

For a single lightweight symbol like EmptyOperator, the plain import is clearly the right call, and that is what is in the PR now after the last commit. I did want to record the reasoning behind the lazier version, since I think it earns its keep once this module grows past one symbol:

• We already repeat the same Airflow 2/3 import dance in many places. The BaseOperator SDK-vs-airflow.models split shows up in 9 modules, and the TaskGroup SDK-vs-airflow.utils.task_group split in 6. Those are exactly the imports this shared compatibility module that you suggested to add here in the PR would own, so we stop copy-pasting the try/except into every file.
• Several symbols behind these try/except guards (not always Airflow2/3, but provider-specific version-aware imports that could be integrated here) are heavy provider imports (Docker, AWS ECS, GCP Cloud Run, Azure Container Instances, BigQuery, KubernetesPod). For those, resolving on first access rather than at module import is a genuine win, since importing the compatibility module would not eagerly pull in every provider.

So the intent was to make this the one place that owns version-aware imports, and during the housekeeping activities fold the scattered try/except blocks into it, with lazy resolution for the heavy ones. The symbols and the justification already exist today, namely the duplications above plus the heavy provider operators, so this is a scope and timing decision rather than waiting for the case to build up. I would like to bring back the lazy form as part of that cleanup.

The eager, class-derived EMPTY_OPERATOR_CLASS_PATH constant is in slight tension with lazy resolution, since computing the dotted path at import forces the class to resolve. If we move heavy operators to lazy access, their path constants would need to be derived lazily too rather than as module-level constants. Easy to handle, just worth keeping in mind so we do not reintroduce eager imports through the path strings.