UPathTools¶

from upathtools import AsyncUPath
import asyncio

async def main():
    # Read a file asynchronously
    path = AsyncUPath("https://example.com/data.json")
    content = await path.aread_text()

    # Iterate directory contents
    async for item in path.aiterdir():
        print(item)

asyncio.run(main())

from upathtools import read_folder
import asyncio

async def main():
    # Read all Python files in parallel
    files = await read_folder(
        "src/",
        pattern="**/*.py",
        load_parallel=True,
        chunk_size=50
    )

    for filepath, content in files.items():
        print(f"{filepath}: {len(content)} bytes")

asyncio.run(main())

from upathtools.filesystems import ModuleFileSystem, ModulePath

# Browse a Python module as a filesystem
fs = ModuleFileSystem(module_name="upathtools")
path: ModulePath = fs.get_upath("async_upath.py")

# Typed info dictionaries
info = path.info()
# info has type-safe fields specific to ModuleFileSystem

from upathtools.filesystems.base import BaseAsyncFileSystem

class MyFileSystem[MyPath, MyInfoDict](BaseAsyncFileSystem):
    """Fully typed filesystem with custom info structure."""

    upath_cls = MyPath

    async def _ls(self, path: str, detail: bool = True) -> list[MyInfoDict]:
        # Return typed info dictionaries
        ...

from upathtools import AsyncUPath
import asyncio

async def main():
    # Read a file asynchronously
    path = AsyncUPath("https://example.com/data.json")
    content = await path.aread_text()

    # Write asynchronously
    output = AsyncUPath("/tmp/output.txt")
    await output.awrite_text("Hello, async!")

    # Iterate directory
    async for item in AsyncUPath("/tmp").aiterdir():
        if await item.ais_file():
            print(item)

asyncio.run(main())

from upathtools import read_folder, list_files
import asyncio

async def main():
    # Read multiple files in parallel
    files = await read_folder(
        "src/",
        pattern="**/*.py",
        load_parallel=True
    )

    for filepath, content in files.items():
        print(f"{filepath}: {len(content)} bytes")

    # Just list files
    py_files = await list_files(
        "src/",
        pattern="**/*.py",
        recursive=True
    )

asyncio.run(main())

from upathtools.filesystems import ModuleFileSystem

# Browse Python modules as filesystems
fs = ModuleFileSystem(module_name="upathtools")

# List module contents
files = fs.ls("/", detail=True)

# Read file with type safety
path = fs.get_upath("async_upath.py")
code = path.read_text()

from upathtools.helpers import to_upath
from pathlib import Path

# Convert any path type to UPath
path = to_upath("/tmp/file.txt")
path = to_upath(Path("/tmp/file.txt"))

# Get async wrapper
async_path = to_upath("/tmp/file.txt", as_async=True)

from upathtools import read_path

async def main():
    # Text mode
    text = await read_path("file.txt", mode="rt")

    # Binary mode
    data = await read_path("image.png", mode="rb")

from upathtools.helpers import write_file

# Handles parent directory creation automatically
write_file("content", "/deep/nested/path/file.txt")
write_file(b"binary", "/path/to/file.bin")

from upathtools.helpers import copy

# Copy between filesystems
copy("s3://bucket/source.txt", "/tmp/local.txt")

# Copy to directory (preserves name)
copy("/tmp/file.txt", "/backup/")

from upathtools.helpers import multi_glob

# Find with multiple patterns
files = multi_glob(
    directory="src/",
    keep_globs=["**/*.py", "**/*.md"],
    drop_globs=["**/test_*.py", "**/__pycache__/**"]
)

from upathtools import read_folder_as_text

async def main():
    # Create single document from multiple files
    combined = await read_folder_as_text(
        "docs/",
        pattern="**/*.md",
        recursive=True
    )
    print(combined)