This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi 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 e4265287cf8 Move `airflow.utils.operator_resources` to task-sdk
(#54461)
e4265287cf8 is described below
commit e4265287cf8829cb19af0464de5015ffd39330c8
Author: Ephraim Anierobi <splendidzig...@gmail.com>
AuthorDate: Fri Aug 15 13:51:55 2025 +0100
Move `airflow.utils.operator_resources` to task-sdk (#54461)
* Move `airflow.utils.operator_resources` to task-sdk
The operator_resources is more of task definition thing. The only usage
in core is during serialization, we can decide how we handle it around
serialization later
* fixup! Move `airflow.utils.operator_resources` to task-sdk
* Apply suggestions from code review
* delete operator_resources
* fixup! delete operator_resources
---
airflow-core/src/airflow/serialization/serialized_objects.py | 2 +-
airflow-core/src/airflow/utils/__init__.py | 3 +++
airflow-core/tests/unit/serialization/test_dag_serialization.py | 2 +-
airflow-core/tests/unit/serialization/test_serialized_objects.py | 2 +-
task-sdk/src/airflow/sdk/bases/operator.py | 6 +++---
task-sdk/src/airflow/sdk/definitions/mappedoperator.py | 2 +-
.../src/airflow/sdk/definitions}/operator_resources.py | 0
.../tests/task_sdk/definitions}/test_operator_resources.py | 2 +-
8 files changed, 11 insertions(+), 8 deletions(-)
diff --git a/airflow-core/src/airflow/serialization/serialized_objects.py
b/airflow-core/src/airflow/serialization/serialized_objects.py
index 997f01ef2fb..0ffa1401275 100644
--- a/airflow-core/src/airflow/serialization/serialized_objects.py
+++ b/airflow-core/src/airflow/serialization/serialized_objects.py
@@ -63,6 +63,7 @@ from airflow.sdk.definitions.asset import (
)
from airflow.sdk.definitions.deadline import DeadlineAlert
from airflow.sdk.definitions.mappedoperator import MappedOperator
+from airflow.sdk.definitions.operator_resources import Resources
from airflow.sdk.definitions.param import Param, ParamsDict
from airflow.sdk.definitions.taskgroup import MappedTaskGroup, TaskGroup
from airflow.sdk.definitions.xcom_arg import serialize_xcom_arg
@@ -93,7 +94,6 @@ from airflow.utils.db import LazySelectSequence
from airflow.utils.docs import get_docs_url
from airflow.utils.log.logging_mixin import LoggingMixin
from airflow.utils.module_loading import import_string, qualname
-from airflow.utils.operator_resources import Resources
from airflow.utils.types import NOTSET, ArgNotSet
if TYPE_CHECKING:
diff --git a/airflow-core/src/airflow/utils/__init__.py
b/airflow-core/src/airflow/utils/__init__.py
index 9ed8b48407c..ab69664df4e 100644
--- a/airflow-core/src/airflow/utils/__init__.py
+++ b/airflow-core/src/airflow/utils/__init__.py
@@ -45,6 +45,9 @@ __deprecated_classes = {
"timeout": {
"timeout": "airflow.sdk.execution_time.timeout.timeout",
},
+ "operator_resources": {
+ "*": "airflow.sdk.definitions.operator_resources",
+ },
}
add_deprecated_classes(__deprecated_classes, __name__)
diff --git a/airflow-core/tests/unit/serialization/test_dag_serialization.py
b/airflow-core/tests/unit/serialization/test_dag_serialization.py
index 132ebc4b994..5020349e594 100644
--- a/airflow-core/tests/unit/serialization/test_dag_serialization.py
+++ b/airflow-core/tests/unit/serialization/test_dag_serialization.py
@@ -68,6 +68,7 @@ from airflow.sdk.bases.decorator import DecoratedOperator
from airflow.sdk.bases.operator import BaseOperator
from airflow.sdk.definitions._internal.expandinput import EXPAND_INPUT_EMPTY
from airflow.sdk.definitions.asset import Asset, AssetUniqueKey
+from airflow.sdk.definitions.operator_resources import Resources
from airflow.sdk.definitions.param import Param, ParamsDict
from airflow.sdk.definitions.taskgroup import TaskGroup
from airflow.security import permissions
@@ -84,7 +85,6 @@ from airflow.ti_deps.deps.ready_to_reschedule import
ReadyToRescheduleDep
from airflow.timetables.simple import NullTimetable, OnceTimetable
from airflow.triggers.base import StartTriggerArgs
from airflow.utils.module_loading import qualname
-from airflow.utils.operator_resources import Resources
from tests_common.test_utils.config import conf_vars
from tests_common.test_utils.markers import
skip_if_force_lowest_dependencies_marker, skip_if_not_on_main
diff --git a/airflow-core/tests/unit/serialization/test_serialized_objects.py
b/airflow-core/tests/unit/serialization/test_serialized_objects.py
index 03f135bd69b..4ac943192c1 100644
--- a/airflow-core/tests/unit/serialization/test_serialized_objects.py
+++ b/airflow-core/tests/unit/serialization/test_serialized_objects.py
@@ -65,6 +65,7 @@ from airflow.sdk.definitions.deadline import (
DeadlineReference,
)
from airflow.sdk.definitions.decorators import task
+from airflow.sdk.definitions.operator_resources import Resources
from airflow.sdk.definitions.param import Param
from airflow.sdk.definitions.taskgroup import TaskGroup
from airflow.sdk.execution_time.context import OutletEventAccessor,
OutletEventAccessors
@@ -78,7 +79,6 @@ from airflow.serialization.serialized_objects import (
from airflow.timetables.base import DataInterval
from airflow.triggers.base import BaseTrigger
from airflow.utils.db import LazySelectSequence
-from airflow.utils.operator_resources import Resources
from airflow.utils.state import DagRunState, State
from airflow.utils.types import DagRunType
diff --git a/task-sdk/src/airflow/sdk/bases/operator.py
b/task-sdk/src/airflow/sdk/bases/operator.py
index c35dc8ada7b..a839d8f89fb 100644
--- a/task-sdk/src/airflow/sdk/bases/operator.py
+++ b/task-sdk/src/airflow/sdk/bases/operator.py
@@ -80,13 +80,13 @@ if TYPE_CHECKING:
from airflow.sdk.bases.operatorlink import BaseOperatorLink
from airflow.sdk.definitions.context import Context
from airflow.sdk.definitions.dag import DAG
+ from airflow.sdk.definitions.operator_resources import Resources
from airflow.sdk.definitions.taskgroup import TaskGroup
from airflow.sdk.definitions.xcom_arg import XComArg
from airflow.serialization.enums import DagAttributeTypes
from airflow.task.priority_strategy import PriorityWeightStrategy
from airflow.triggers.base import BaseTrigger, StartTriggerArgs
from airflow.typing_compat import Self
- from airflow.utils.operator_resources import Resources
TaskPreExecuteHook = Callable[[Context], None]
TaskPostExecuteHook = Callable[[Context, Any], None]
@@ -177,7 +177,7 @@ def coerce_timedelta(value: float | timedelta, *, key: str
| None = None) -> tim
def coerce_resources(resources: dict[str, Any] | None) -> Resources | None:
if resources is None:
return None
- from airflow.utils.operator_resources import Resources
+ from airflow.sdk.definitions.operator_resources import Resources
return Resources(**resources)
@@ -1330,7 +1330,7 @@ class BaseOperator(AbstractOperator,
metaclass=BaseOperatorMeta):
if resources is None:
return None
- from airflow.utils.operator_resources import Resources
+ from airflow.sdk.definitions.operator_resources import Resources
if isinstance(resources, Resources):
return resources
diff --git a/task-sdk/src/airflow/sdk/definitions/mappedoperator.py
b/task-sdk/src/airflow/sdk/definitions/mappedoperator.py
index ba5595e3b48..260a4b57a59 100644
--- a/task-sdk/src/airflow/sdk/definitions/mappedoperator.py
+++ b/task-sdk/src/airflow/sdk/definitions/mappedoperator.py
@@ -66,9 +66,9 @@ if TYPE_CHECKING:
)
from airflow.sdk import DAG, BaseOperator, BaseOperatorLink, Context,
TaskGroup, XComArg
from airflow.sdk.definitions._internal.expandinput import ExpandInput
+ from airflow.sdk.definitions.operator_resources import Resources
from airflow.sdk.definitions.param import ParamsDict
from airflow.triggers.base import StartTriggerArgs
- from airflow.utils.operator_resources import Resources
from airflow.utils.trigger_rule import TriggerRule
ValidationSource = Literal["expand"] | Literal["partial"]
diff --git a/airflow-core/src/airflow/utils/operator_resources.py
b/task-sdk/src/airflow/sdk/definitions/operator_resources.py
similarity index 100%
rename from airflow-core/src/airflow/utils/operator_resources.py
rename to task-sdk/src/airflow/sdk/definitions/operator_resources.py
diff --git a/airflow-core/tests/unit/utils/test_operator_resources.py
b/task-sdk/tests/task_sdk/definitions/test_operator_resources.py
similarity index 95%
rename from airflow-core/tests/unit/utils/test_operator_resources.py
rename to task-sdk/tests/task_sdk/definitions/test_operator_resources.py
index fa25106017d..9e0875cf076 100644
--- a/airflow-core/tests/unit/utils/test_operator_resources.py
+++ b/task-sdk/tests/task_sdk/definitions/test_operator_resources.py
@@ -17,7 +17,7 @@
# under the License.
from __future__ import annotations
-from airflow.utils.operator_resources import Resources
+from airflow.sdk.definitions.operator_resources import Resources
class TestResources:
