From c85676d2340cfde49776b2bf7e3dd86e2bad8751 Mon Sep 17 00:00:00 2001
From: Pavan Kumar <gopidesupavan@gmail.com>
Date: Tue, 30 Sep 2025 22:00:04 +0100
Subject: [PATCH 1/4] Fix ObjectStoragePath compatible with universal-pathlib
 0.3.0 to avoid maximum recursion depth exceeded

---
 task-sdk/src/airflow/sdk/io/path.py     | 4 ++--
 task-sdk/tests/task_sdk/io/test_path.py | 4 ++--
 2 files changed, 4 insertions(+), 4 deletions(-)

diff --git a/task-sdk/src/airflow/sdk/io/path.py b/task-sdk/src/airflow/sdk/io/path.py
index 5b66e09d110a0..7be19e9b70e49 100644
--- a/task-sdk/src/airflow/sdk/io/path.py
+++ b/task-sdk/src/airflow/sdk/io/path.py
@@ -294,7 +294,7 @@ def _cp_file(self, dst: ObjectStoragePath, **kwargs):
             # make use of system dependent buffer size
             shutil.copyfileobj(f1, f2, **kwargs)
 
-    def copy(self, dst: str | ObjectStoragePath, recursive: bool = False, **kwargs) -> None:
+    def copy(self, dst: str | ObjectStoragePath, recursive: bool = False, **kwargs) -> None:  # type: ignore[override]
         """
         Copy file(s) from this path to another location.
 
@@ -413,5 +413,5 @@ def deserialize(cls, data: dict, version: int) -> ObjectStoragePath:
     def __str__(self):
         conn_id = self.storage_options.get("conn_id")
         if self._protocol and conn_id:
-            return f"{self._protocol}://{conn_id}@{self.path}"
+            return f"{self._protocol}://{conn_id}@{self.parser.join(*self._raw_urlpaths)}"
         return super().__str__()
diff --git a/task-sdk/tests/task_sdk/io/test_path.py b/task-sdk/tests/task_sdk/io/test_path.py
index 78c75cebddf6e..11880d727171b 100644
--- a/task-sdk/tests/task_sdk/io/test_path.py
+++ b/task-sdk/tests/task_sdk/io/test_path.py
@@ -102,7 +102,7 @@ def _strip_protocol(cls, path):
 class TestAttach:
     FAKE = "ffs:///fake"
     MNT = "ffs:///mnt/warehouse"
-    FOO = "ffs:///mnt/warehouse/foo"
+    FOO = "ffs://fake@/mnt/warehouse/foo"
     BAR = FOO
 
     @pytest.fixture(autouse=True)
@@ -253,7 +253,7 @@ def test_relative_to(self, tmp_path, target):
         o1 = ObjectStoragePath(f"file://{target}")
         o2 = ObjectStoragePath(f"file://{tmp_path.as_posix()}")
         o3 = ObjectStoragePath(f"file:///{uuid.uuid4()}")
-        assert o1.relative_to(o2) == o1
+        assert o1.is_relative_to(o2)
         with pytest.raises(ValueError, match="is not in the subpath of"):
             o1.relative_to(o3)
 

From d9953f450e28eb816cc2998c122daf0be3801361 Mon Sep 17 00:00:00 2001
From: Pavan Kumar <gopidesupavan@gmail.com>
Date: Tue, 30 Sep 2025 22:57:42 +0100
Subject: [PATCH 2/4] Attemp mypy fix

---
 airflow-core/pyproject.toml | 4 +---
 1 file changed, 1 insertion(+), 3 deletions(-)

diff --git a/airflow-core/pyproject.toml b/airflow-core/pyproject.toml
index 0dfe6ed78b97d..7d6f629f4f16b 100644
--- a/airflow-core/pyproject.toml
+++ b/airflow-core/pyproject.toml
@@ -134,9 +134,7 @@ dependencies = [
     "tenacity>=8.3.0",
     "termcolor>=3.0.0",
     "typing-extensions>=4.14.1",
-    # Universal Pathlib 0.2.4 adds extra validation for Paths and our integration with local file paths
-    # Does not work with it Tracked in https://github.com/fsspec/universal_pathlib/issues/276
-    "universal-pathlib>=0.2.2,!=0.2.4",
+    "universal-pathlib>=0.3.0",
     "uuid6>=2024.7.10",
     "apache-airflow-task-sdk<1.3.0,>=1.1.0",
     # pre-installed providers

From 4621d9fb9519f5215453ebd83d1fbbd3062fb323 Mon Sep 17 00:00:00 2001
From: Pavan Kumar <gopidesupavan@gmail.com>
Date: Fri, 3 Oct 2025 07:47:30 +0100
Subject: [PATCH 3/4] Update test_relative_to test assertion

---
 task-sdk/tests/task_sdk/io/test_path.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/task-sdk/tests/task_sdk/io/test_path.py b/task-sdk/tests/task_sdk/io/test_path.py
index 11880d727171b..d367ebb537500 100644
--- a/task-sdk/tests/task_sdk/io/test_path.py
+++ b/task-sdk/tests/task_sdk/io/test_path.py
@@ -253,7 +253,7 @@ def test_relative_to(self, tmp_path, target):
         o1 = ObjectStoragePath(f"file://{target}")
         o2 = ObjectStoragePath(f"file://{tmp_path.as_posix()}")
         o3 = ObjectStoragePath(f"file:///{uuid.uuid4()}")
-        assert o1.is_relative_to(o2)
+        assert o1.relative_to(o2)
         with pytest.raises(ValueError, match="is not in the subpath of"):
             o1.relative_to(o3)
 

From 84b083df370da61d2961a0dc02dc6cff0eb7a13d Mon Sep 17 00:00:00 2001
From: Pavan Kumar <gopidesupavan@gmail.com>
Date: Fri, 3 Oct 2025 09:27:28 +0100
Subject: [PATCH 4/4] Add todo comment for type ignore

---
 task-sdk/src/airflow/sdk/io/path.py | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/task-sdk/src/airflow/sdk/io/path.py b/task-sdk/src/airflow/sdk/io/path.py
index 7be19e9b70e49..8023c2324869c 100644
--- a/task-sdk/src/airflow/sdk/io/path.py
+++ b/task-sdk/src/airflow/sdk/io/path.py
@@ -307,6 +307,8 @@ def copy(self, dst: str | ObjectStoragePath, recursive: bool = False, **kwargs)
 
         kwargs: Additional keyword arguments to be passed to the underlying implementation.
         """
+        # TODO: change dst arg to target when we bump major version for task-sdk and remove type: ignore in copy method
+
         from airflow.lineage.hook import get_hook_lineage_collector
 
         if isinstance(dst, str):
