Repository: apache/airflow
File: task-sdk/src/airflow/sdk/io/path.py (Python)

Code before:
        protocol: str | None = None,
        conn_id: str | None = None,
        **storage_options: Any,
    ) -> None:
        # ensure conn_id is always set in storage_options
        storage_options.setdefault("conn_id", None)
        # parse conn_id from args if provided
        if args:
            arg0 = args[0]
            if isinstance(arg0, type(self)):
                storage_options["conn_id"] = arg0.conn_id
            else:
                parsed_url = urlsplit(stringify_path(arg0))
                userinfo, have_info, hostinfo = parsed_url.netloc.rpartition("@")
                if have_info:
                    conn_id = storage_options["conn_id"] = userinfo or None
                    parsed_url = parsed_url._replace(netloc=hostinfo)
                args = (parsed_url.geturl(),) + args[1:]
                protocol = protocol or parsed_url.scheme
        # override conn_id if explicitly provided
        if conn_id is not None:
            storage_options["conn_id"] = conn_id

        # pop conn_id before calling super to prevent it from being passed
        # to the underlying fsspec filesystem, which doesn't understand it
        self._conn_id = storage_options.pop("conn_id", None)
        super().__init__(*args, protocol=protocol, **storage_options)

    @property
    def conn_id(self) -> str | None:
        """Return the connection ID for this path."""
        return getattr(self, "_conn_id", None)

    @property
    def fs(self) -> AbstractFileSystem:
        """Return the filesystem for this path, using airflow's attach mechanism."""
        return attach(self.protocol or "file", self.conn_id).fs

    def __hash__(self) -> int:
        self._hash_cached: int
        try:
            return self._hash_cached
        except AttributeError:
            self._hash_cached = hash(str(self))
            return self._hash_cached

    def __eq__(self, other: Any) -> bool:
        return self.samestore(other) and str(self) == str(other)

    def samestore(self, other: Any) -> bool:
        return (

Review comment (bug):
Can we also add a regression test for path derivation here?

1) `ObjectStoragePath("fake://my_conn@bucket/key") / "sub"` keeps `conn_id == "my_conn"` and stringifies with `my_conn@...`
2) `conn_id` is still not passed to the fsspec filesystem constructor.

That should cover both the original bug and propagation regressions.

Code after:
        protocol: str | None = None,
        conn_id: str | None = None,
        **storage_options: Any,
    ) -> None:
        # ensure conn_id is always set in storage_options
        storage_options.setdefault("conn_id", None)
        # parse conn_id from args if provided
        if args:
            arg0 = args[0]
            if isinstance(arg0, type(self)):
                storage_options["conn_id"] = arg0.conn_id
            else:
                parsed_url = urlsplit(stringify_path(arg0))
                userinfo, have_info, hostinfo = parsed_url.netloc.rpartition("@")
                if have_info:
                    conn_id = storage_options["conn_id"] = userinfo or None
                    parsed_url = parsed_url._replace(netloc=hostinfo)
                args = (parsed_url.geturl(),) + args[1:]
                protocol = protocol or parsed_url.scheme
        # override conn_id if explicitly provided
        if conn_id is not None:
            storage_options["conn_id"] = conn_id

        # pop conn_id before calling super to prevent it from being passed
        # to the underlying fsspec filesystem, which doesn't understand it
        self._conn_id = storage_options.pop("conn_id", None)
        super().__init__(*args, protocol=protocol, **storage_options)

    @classmethod_or_method  # type: ignore[arg-type]
    def _from_upath(cls_or_self, upath, /):
        """Wrap a UPath, propagating conn_id from the calling instance."""
        is_instance = isinstance(cls_or_self, ObjectStoragePath)
        cls = type(cls_or_self) if is_instance else cls_or_self
        if isinstance(upath, cls):
            return upath
        obj = object.__new__(cls)
        obj.__wrapped__ = upath
        obj._conn_id = getattr(cls_or_self, "_conn_id", None) if is_instance else None
        return obj

    @property
    def conn_id(self) -> str | None:
        """Return the connection ID for this path."""
        return getattr(self, "_conn_id", None)

    @property
    def fs(self) -> AbstractFileSystem:
        """Return the filesystem for this path, using airflow's attach mechanism."""
        return attach(self.protocol or "file", self.conn_id).fs

    def __hash__(self) -> int: