Skip to content

Update BaseOperator imports for Airflow 3.0 compatibility #52378

New issue
New issue
Closed
Closed
Update BaseOperator imports for Airflow 3.0 compatibility#52378
Assignees
gylibdsohakevinhongzlkyungjunleemekgw7401sjyangkevinjschefflhardeybiseyjroachgolf84karenbraganz
Labels
area:providerskind:featureFeature Requestskind:metaHigh-level information important to the community
@kaxil

Description

@kaxil
kaxil
opened on Jun 28, 2025

We need community help to migrate all provider packages to use the new Task SDK BaseOperator import for Airflow 3.0 compatibility while maintaining backward compatibility with Airflow 2.x.

Background

With Airflow 3.0, the BaseOperator has moved to the Task SDK (airflow.sdk). All provider packages need to be updated to use version-compatible imports to work with both Airflow 2.x and 3.x versions.

Migration Tasks

Use #52366 & #52292 as reference.

Each provider needs these updates:

  1. Create/update version_compat.py with standardized version compatibility pattern
  2. Update imports in all operator, sensor, hook, and link files
  3. Fix BaseOperatorLink.persist methods that cause AttributeError with SDK BaseOperator
  4. Update test patterns to use dag_maker.run_ti() instead of deprecated patterns

Provider Status Checklist

Implementation Guide

If using AI Agents, refer to #52378 (comment)

Step 1: Create version_compat.py

# In providers/{provider_name}/src/airflow/providers/{provider_name}/version_compat.py
from __future__ import annotations

def get_base_airflow_version_tuple() -> tuple[int, int, int]:
    from packaging.version import Version
    from airflow import __version__
    airflow_version = Version(__version__)
    return airflow_version.major, airflow_version.minor, airflow_version.micro

AIRFLOW_V_3_0_PLUS = get_base_airflow_version_tuple() >= (3, 0, 0)

if AIRFLOW_V_3_0_PLUS:
    from airflow.sdk import BaseOperator
else:
    from airflow.models import BaseOperator

__all__ = ["AIRFLOW_V_3_0_PLUS", "BaseOperator"]

Step 2: Update imports

Replace:

  from airflow.models.baseoperator import BaseOperator

With:

  from airflow.providers.{provider_name}.version_compat import BaseOperator

Step 3: Fix BaseOperatorLink.persist methods

Replace:

  def persist(context: Context, task_instance: BaseOperator, url: str):
      task_instance.xcom_push(key="link_key", value=url)

With:

  def persist(context: Context, url: str):
      context["task_instance"].xcom_push(key="link_key", value=url)

Step 4: Update test patterns

Replace operator.run() with dag_maker.run_ti(task_id, dr). Use #52366 & #52292 as reference.

Test your changes:

  # Run provider tests
  breeze testing providers-tests --test-type "Providers[{provider_name}]"

How to Contribute

  1. Pick an unchecked provider from the list above
  2. Comment on this issue to claim it (prevents duplicate work)
  3. Follow the implementation guide
  4. Submit a PR with title: Provider Migration: Update {provider_name} for Airflow 3.0 compatibility
  5. Check off the completed provider in this issue

Metadata

Metadata

Assignees

Labels

area:providerskind:featureFeature Requestskind:metaHigh-level information important to the community

Type

No type

Projects

No projects

Milestone

No milestone

Relationships

None yet

Development

No branches or pull requests

Issue actions