async_read

Class info

🛈 DocStrings

_get_cached_fs cached

_get_cached_fs(protocol: str) -> AsyncFileSystem

Cached filesystem creation.

Source code in src/llmling_agent/utils/async_read.py

@lru_cache(maxsize=32)
def _get_cached_fs(protocol: str) -> AsyncFileSystem:
    """Cached filesystem creation."""
    import fsspec
    from fsspec.asyn import AsyncFileSystem
    from fsspec.implementations.asyn_wrapper import AsyncFileSystemWrapper
    from morefs.asyn_local import AsyncLocalFileSystem

    if protocol in ("", "file"):
        return AsyncLocalFileSystem()

    fs = fsspec.filesystem(protocol, asynchronous=True)
    if not isinstance(fs, AsyncFileSystem):
        fs = AsyncFileSystemWrapper(fs)
    return fs

get_async_fs async

get_async_fs(path: StrPath) -> AsyncFileSystem

Get appropriate async filesystem for path.

Source code in src/llmling_agent/utils/async_read.py

async def get_async_fs(path: StrPath) -> AsyncFileSystem:
    """Get appropriate async filesystem for path."""
    from upath import UPath

    path_obj = UPath(path)
    return _get_cached_fs(path_obj.protocol)

read_path async

read_path(path: StrPath, mode: Literal['rt'] = 'rt', encoding: str = ...) -> str

read_path(path: StrPath, mode: Literal['rb'], encoding: str = ...) -> bytes

read_path(
    path: StrPath, mode: Literal["rt", "rb"] = "rt", encoding: str = "utf-8"
) -> str | bytes

Read file content asynchronously when possible.

Parameters:

Name	Type	Description	Default
path	StrPath	Path to read	required
mode	Literal['rt', 'rb']	Read mode ("rt" for text, "rb" for binary)	'rt'
encoding	str	File encoding for text files	'utf-8'

Returns:

Type	Description
str | bytes	File content as string or bytes depending on mode

Source code in src/llmling_agent/utils/async_read.py

async def read_path(
    path: StrPath,
    mode: Literal["rt", "rb"] = "rt",
    encoding: str = "utf-8",
) -> str | bytes:
    """Read file content asynchronously when possible.

    Args:
        path: Path to read
        mode: Read mode ("rt" for text, "rb" for binary)
        encoding: File encoding for text files

    Returns:
        File content as string or bytes depending on mode
    """
    from upath import UPath

    path_obj = UPath(path)
    fs = await get_async_fs(path_obj)

    f = await fs.open_async(path_obj.path, mode=mode)
    async with f:
        return await f.read()