This is an automated email from the ASF dual-hosted git repository.
shahar1 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 5a1f6ed0239 Keep DualStatsManager as a try/except import as it is soon
going to be removed (#65676)
5a1f6ed0239 is described below
commit 5a1f6ed023941949a7cc924371e6d91811c8093d
Author: Ash Berlin-Taylor <[email protected]>
AuthorDate: Wed Apr 22 21:05:10 2026 +0100
Keep DualStatsManager as a try/except import as it is soon going to be
removed (#65676)
---
.../edge3/src/airflow/providers/edge3/models/edge_worker.py | 10 ++++++----
.../src/airflow/providers/edge3/worker_api/routes/jobs.py | 13 ++++++-------
.../src/airflow/providers/edge3/worker_api/routes/worker.py | 10 ++++++----
3 files changed, 18 insertions(+), 15 deletions(-)
diff --git a/providers/edge3/src/airflow/providers/edge3/models/edge_worker.py
b/providers/edge3/src/airflow/providers/edge3/models/edge_worker.py
index 5f2f7e999e5..a294cd4b44a 100644
--- a/providers/edge3/src/airflow/providers/edge3/models/edge_worker.py
+++ b/providers/edge3/src/airflow/providers/edge3/models/edge_worker.py
@@ -28,12 +28,16 @@ from sqlalchemy.orm import Mapped
from airflow.providers.common.compat.sdk import AirflowException, timezone
from airflow.providers.common.compat.sqlalchemy.orm import mapped_column
from airflow.providers.edge3.models.edge_base import Base
-from airflow.providers.edge3.version_compat import AIRFLOW_V_3_2_PLUS
from airflow.utils.log.logging_mixin import LoggingMixin
from airflow.utils.providers_configuration_loader import
providers_configuration_loaded
from airflow.utils.session import NEW_SESSION, provide_session
from airflow.utils.sqlalchemy import UtcDateTime
+try:
+ from airflow.sdk.observability.stats import DualStatsManager
+except ImportError:
+ DualStatsManager = None # type: ignore[assignment,misc] # Airflow <
3.2.1 compat
+
if TYPE_CHECKING:
from collections.abc import Sequence
@@ -180,9 +184,7 @@ def set_metrics(
"free_concurrency",
}
- if AIRFLOW_V_3_2_PLUS:
- from airflow.sdk.observability.stats import DualStatsManager
-
+ if DualStatsManager is not None:
try:
DualStatsManager.gauge(
"edge_worker.status",
diff --git
a/providers/edge3/src/airflow/providers/edge3/worker_api/routes/jobs.py
b/providers/edge3/src/airflow/providers/edge3/worker_api/routes/jobs.py
index e18fed366ac..0ecf9952024 100644
--- a/providers/edge3/src/airflow/providers/edge3/worker_api/routes/jobs.py
+++ b/providers/edge3/src/airflow/providers/edge3/worker_api/routes/jobs.py
@@ -22,13 +22,16 @@ from typing import Annotated
from fastapi import Body, Depends, status
from sqlalchemy import select, update
+try:
+ from airflow.sdk.observability.stats import DualStatsManager
+except ImportError:
+ DualStatsManager = None # type: ignore[assignment,misc] # Airflow < 3.2
compat
from airflow.api_fastapi.common.db.common import SessionDep # noqa: TC001
from airflow.api_fastapi.common.router import AirflowRouter
from airflow.api_fastapi.core_api.openapi.exceptions import
create_openapi_http_exception_doc
from airflow.executors.workloads import ExecuteTask
from airflow.providers.common.compat.sdk import timezone
from airflow.providers.edge3.models.edge_job import EdgeJobModel
-from airflow.providers.edge3.version_compat import AIRFLOW_V_3_2_PLUS
from airflow.providers.edge3.worker_api.auth import
jwt_token_authorization_rest
from airflow.providers.edge3.worker_api.datamodels import (
EdgeJobFetched,
@@ -88,9 +91,7 @@ def fetch(
session.commit()
# Edge worker does not backport emitted Airflow metrics, so export some
metrics
tags = {"dag_id": job.dag_id, "task_id": job.task_id, "queue": job.queue}
- if AIRFLOW_V_3_2_PLUS:
- from airflow.sdk.observability.stats import DualStatsManager
-
+ if DualStatsManager is not None:
DualStatsManager.incr("edge_worker.ti.start", tags=tags)
else:
from airflow.providers.common.compat.sdk import Stats
@@ -149,9 +150,7 @@ def state(
"queue": job.queue,
"state": str(state),
}
- if AIRFLOW_V_3_2_PLUS:
- from airflow.sdk.observability.stats import DualStatsManager
-
+ if DualStatsManager is not None:
DualStatsManager.incr(
"edge_worker.ti.finish",
tags=tags,
diff --git
a/providers/edge3/src/airflow/providers/edge3/worker_api/routes/worker.py
b/providers/edge3/src/airflow/providers/edge3/worker_api/routes/worker.py
index 223c94afb73..81d54963605 100644
--- a/providers/edge3/src/airflow/providers/edge3/worker_api/routes/worker.py
+++ b/providers/edge3/src/airflow/providers/edge3/worker_api/routes/worker.py
@@ -28,7 +28,6 @@ from airflow.api_fastapi.common.router import AirflowRouter
from airflow.api_fastapi.core_api.openapi.exceptions import
create_openapi_http_exception_doc
from airflow.providers.common.compat.sdk import timezone
from airflow.providers.edge3.models.edge_worker import EdgeWorkerModel,
EdgeWorkerState, set_metrics
-from airflow.providers.edge3.version_compat import AIRFLOW_V_3_2_PLUS
from airflow.providers.edge3.worker_api.auth import
jwt_token_authorization_rest
from airflow.providers.edge3.worker_api.datamodels import (
WorkerQueueUpdateBody,
@@ -37,6 +36,11 @@ from airflow.providers.edge3.worker_api.datamodels import (
WorkerStateBody,
)
+try:
+ from airflow.sdk.observability.stats import DualStatsManager
+except ImportError:
+ DualStatsManager = None # type: ignore[assignment,misc] # Airflow < 3.2
compat
+
worker_router = AirflowRouter(
tags=["Worker"],
prefix="/worker",
@@ -217,9 +221,7 @@ def set_state(
worker.sysinfo = body.sysinfo
worker.last_update = timezone.utcnow()
session.commit()
- if AIRFLOW_V_3_2_PLUS:
- from airflow.sdk.observability.stats import DualStatsManager
-
+ if DualStatsManager is not None:
DualStatsManager.incr(
"edge_worker.heartbeat_count",
1,
