Add extra

airflow/io/store/path.py

@@ -45,6 +45,13 @@ class ObjectStoragePath(os.PathLike):
     sep: typing.ClassVar[str] = "/"
     root_marker: typing.ClassVar[str] = "/"
 
+    _store: ObjectStore | None
+    _bucket: str
+    _key: str
+    _conn_id: str | None
+    _protocol: str
+    _hash: int | None
+
     __slots__ = (
         "_store",
         "_bucket",
@@ -54,18 +61,26 @@ class ObjectStoragePath(os.PathLike):
         "_hash",
     )
 
-    def __init__(self, path, conn_id: str | None = None, store: ObjectStore | None = None):
+    def __init__(
+        self, path: str | ObjectStoragePath, conn_id: str | None = None, store: ObjectStore | None = None
+    ):
         self._conn_id = conn_id
         self._store = store
 
         self._hash = None
 
-        self._protocol, self._bucket, self._key = self.split_path(path)
+        if isinstance(path, ObjectStoragePath):
+            self._protocol = path._protocol
+            self._bucket = path._bucket
+            self._key = path._key
+            self._store = path._store
+        else:
+            self._protocol, self._bucket, self._key = self.split_path(path)
 
         if store:
             self._conn_id = store.conn_id
             self._protocol = self._protocol if self._protocol else store.protocol
-        elif self._protocol:
+        elif self._protocol and not self._store:
             self._store = attach(self._protocol, conn_id)
 
     @classmethod

tests/io/store/test_store.py

@@ -61,6 +61,11 @@ def test_init_objectstoragepath(self):
         assert path.key == "key/part1/part2"
         assert path._protocol == "file"
 
+        path2 = ObjectStoragePath(path / "part3")
+        assert path2.bucket == "bucket"
+        assert path2.key == "key/part1/part2/part3"
+        assert path2._protocol == "file"
+
     def test_read_write(self):
         o = ObjectStoragePath(f"file:///tmp/{str(uuid.uuid4())}")