Migrate redis provider to common.compat

[kaxil]

Replace version-specific conditional imports with common.compat layer. This standardizes compatibility handling across Airflow 2.x and 3.x.

---

^ 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.

[potiuk]

Compat failing on missing next_dagrun_info ?

[jason810496]

Compat failing on missing next_dagrun_info ?

Yes, since after the change we will import airflow.sdk.DAG for AIRFLOW_V_3_0_PLUS, but we need to import airflow.models.DAG to make the following Dag setup works.

airflow/providers/redis/tests/unit/redis/log/test_redis_task_handler.py

Lines 85 to 96 in f3ad20a

	bundle_name = "testing"
	if AIRFLOW_V_3_1_PLUS:
	sync_dag_to_db(dag, bundle_name=bundle_name, session=session)
	elif AIRFLOW_V_3_0_PLUS:
	from airflow.models.dagbundle import DagBundleModel
	from airflow.models.serialized_dag import SerializedDagModel
	from airflow.serialization.serialized_objects import SerializedDAG
	session.add(DagBundleModel(name=bundle_name))
	session.flush()
	SerializedDAG.bulk_write_to_db(bundle_name, None, [dag])
	SerializedDagModel.write_dag(dag, bundle_name=bundle_name)

So maybe we could rewrite the test to just setup the TI instead of the whole Dag like TestGCSTaskHandler one:

airflow/providers/google/tests/unit/google/cloud/log/test_gcs_task_handler.py

Lines 39 to 52 in f3ad20a

	def task_instance(self, create_task_instance, session):
	self.ti = ti = create_task_instance(
	dag_id="dag_for_testing_gcs_task_handler",
	task_id="task_for_testing_gcs_task_handler",
	logical_date=datetime(2020, 1, 1),
	state=TaskInstanceState.RUNNING,
	)
	ti.try_number = 1
	ti.raw = False
	session.add(ti)
	session.commit()
	yield
	clear_db_runs()
	clear_db_dags()

[jason810496]

I can help out rewrite the test tomorrow if this is not very urgent

[kaxil]

This works. Failure is unrelated -- its a transient mongo error