Class AsyncPath

marimo/_utils/async_path.py:21–218

An async version of pathlib.Path that uses asyncio.to_thread for filesystem operations. This class inherits from PurePath for path manipulation and adds async filesystem methods.

class AsyncPath(PurePath):
    """
    An async version of pathlib.Path that uses asyncio.to_thread for filesystem operations.

    This class inherits from PurePath for path manipulation and adds async filesystem methods.
    """

    def __new__(cls, *args: Any, **kwargs: Any) -> Self:
        # Create the path using the same logic as PurePath
        path_cls: type[AsyncPath]
        if cls is AsyncPath:
            path_cls = AsyncWindowsPath if os.name == "nt" else AsyncPosixPath
        else:
            path_cls = cls
        return super().__new__(path_cls, *args, **kwargs)  # type: ignore

    def __truediv__(self, other: StrPath) -> AsyncPath:
        # Override to return AsyncPath instance
        result = super().__truediv__(other)
        return self.__class__(result)

    def __rtruediv__(self, other: StrPath) -> AsyncPath:
        # Override to return AsyncPath instance
        result = super().__rtruediv__(other)
        return self.__class__(result)

    @property
    def _path(self) -> Path:
        """Get the synchronous Path equivalent."""
        return Path(self)

    # Async filesystem operations

    async def exists(self) -> bool:
        """Return True if the path exists."""
        return await asyncio.to_thread(self._path.exists)

    async def is_file(self) -> bool:
        """Return True if the path is a regular file."""
        return await asyncio.to_thread(self._path.is_file)

    async def is_dir(self) -> bool:
        """Return True if the path is a directory."""
        return await asyncio.to_thread(self._path.is_dir)

    async def is_symlink(self) -> bool:
        """Return True if the path is a symbolic link."""
        return await asyncio.to_thread(self._path.is_symlink)

    async def stat(self) -> os.stat_result:
        """Return stat info for the path."""
        return await asyncio.to_thread(self._path.stat)

    async def lstat(self) -> os.stat_result:
        """Return lstat info for the path (doesn't follow symlinks)."""
        return await asyncio.to_thread(self._path.lstat)

    async def chmod(self, mode: int) -> None: