tessl/pypi-apache-airflow-providers-common-compat
Common Compatibility Provider - providing compatibility code for previous Airflow versions
—
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
version-compatibility.mddocs/
Version Compatibility
Core utilities for detecting Airflow versions and importing version-appropriate modules. This module provides runtime detection of Airflow versions and conditional imports to ensure compatibility across different Airflow releases.
Capabilities
Version Detection
Returns the base Airflow version as a tuple of major, minor, and micro version numbers for programmatic version comparison.
def get_base_airflow_version_tuple() -> tuple[int, int, int]:
"""
Get the base Airflow version as a tuple.
Returns:
tuple[int, int, int]: Major, minor, micro version numbers
"""Version Constants
Boolean constants that indicate whether the current Airflow installation meets specific version thresholds.
AIRFLOW_V_3_0_PLUS: bool # True if Airflow version is 3.0+
AIRFLOW_V_3_1_PLUS: bool # True if Airflow version is 3.1+Compatible BaseOperator
Version-compatible BaseOperator class that imports from the appropriate module based on the Airflow version.
class BaseOperator:
"""
Version-compatible BaseOperator class.
Imports from airflow.sdk.BaseOperator for Airflow 3.0+
Imports from airflow.models.BaseOperator for Airflow < 3.0
"""Usage Examples
from airflow.providers.common.compat.version_compat import (
get_base_airflow_version_tuple,
AIRFLOW_V_3_0_PLUS,
AIRFLOW_V_3_1_PLUS,
BaseOperator
)
# Check specific version
version = get_base_airflow_version_tuple()
if version >= (2, 10, 0):
print("Compatible with this provider")
# Use version flags for conditional logic
if AIRFLOW_V_3_0_PLUS:
# Use Airflow 3.0+ features
from airflow.sdk import SomeNewFeature
else:
# Use legacy Airflow features
from airflow.models import SomeLegacyFeature
# Create operators using compatible BaseOperator
class MyOperator(BaseOperator):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
def execute(self, context):
# Implementation works across versions
passInstall with Tessl CLI
npx tessl i tessl/pypi-apache-airflow-providers-common-compat