AsyncFileSystem | Daytona

AsyncFileSystem

class AsyncFileSystem()

Provides file system operations within a Sandbox.

This class implements a high-level interface to file system operations that can be performed within a Daytona Sandbox.

AsyncFileSystem.init

def __init__(api_client: FileSystemApi)

Initializes a new FileSystem instance.

Arguments:

AsyncFileSystem.create_folder

@intercept_errors(message_prefix="Failed to create folder: ")
@with_instrumentation()
async def create_folder(path: str,
                        mode: str,
                        request_timeout: float | None = None) -> None

Creates a new directory in the Sandbox at the specified path with the given permissions.

Arguments:

Example:

# Create a directory with standard permissions
await sandbox.fs.create_folder("workspace/data", "755")

# Create a private directory
await sandbox.fs.create_folder("workspace/secrets", "700")

AsyncFileSystem.delete_file

@intercept_errors(message_prefix="Failed to delete file: ")
@with_instrumentation()
async def delete_file(path: str,
                      recursive: bool = False,
                      request_timeout: float | None = None) -> None

Deletes a file from the Sandbox.

Arguments:

Example:

# Delete a file
await sandbox.fs.delete_file("workspace/data/old_file.txt")

AsyncFileSystem.download_file

@overload
async def download_file(remote_path: str, timeout: int = 30 * 60) -> bytes

Downloads a file from the Sandbox. Returns the file contents as a bytes object. This method is useful when you want to load the file into memory without saving it to disk. It can only be used for smaller files.

Arguments:

Returns:

Example:

# Download and save a file locally
content = await sandbox.fs.download_file("workspace/data/file.txt")

with open("local_copy.txt", "wb") as f:
    f.write(content)

# Download and process text content
content = await sandbox.fs.download_file("workspace/data/config.json")
config = json.loads(content.decode('utf-8'))

AsyncFileSystem.download_file_stream

@intercept_errors(message_prefix="Failed to download file: ")
@with_instrumentation()
async def download_file_stream(
        remote_path: str,
        timeout: int = 30 * 60,
        on_progress: Callable[[DownloadProgress], Awaitable[None] | None]
    | None = None,
        cancel_event: CancelEvent | None = None) -> AsyncIterator[bytes]

Downloads a single file from the Sandbox as a stream without buffering the entire file into memory. Returns an async iterator that yields file content in chunks, which can be piped directly to an HTTP response, written to a file incrementally, or processed on the fly.

Arguments:

Returns:

Raises:

Example:

# Stream to a local file without loading into memory
async with aiofiles.open("local_copy.bin", "wb") as f:
    async for chunk in await sandbox.fs.download_file_stream("workspace/large-file.bin"):
        await f.write(chunk)

# Cancel a download from another coroutine
import asyncio
cancel = asyncio.Event()
asyncio.get_running_loop().call_later(5.0, cancel.set)
async for chunk in await sandbox.fs.download_file_stream("workspace/big.bin", cancel_event=cancel):
    process(chunk)

AsyncFileSystem.list_files

@intercept_errors(message_prefix="Failed to list files: ")
@with_instrumentation()
async def list_files(path: str,
                     depth: int | None = None,
                     request_timeout: float | None = None) -> list[FileInfo]

Lists files and directories in a given path and returns their information, similar to the ls -l command.

Arguments:

Returns:

Example:

# List directory contents
files = await sandbox.fs.list_files("workspace/data")
# Print files and their sizes
for file in files:
    if not file.is_dir:
        print(f"{file.name}: {file.size} bytes")
# List recursively two levels deep
tree = await sandbox.fs.list_files("workspace/data", depth=2)
print("Paths:", ", ".join(f.path for f in tree))

AsyncFileSystem.move_files

@intercept_errors(message_prefix="Failed to move files: ")
@with_instrumentation()
async def move_files(source: str,
                     destination: str,
                     request_timeout: float | None = None) -> None

Moves or renames a file or directory. The parent directory of the destination must exist.

Arguments:

Example:

# Rename a file
await sandbox.fs.move_files(
    "workspace/data/old_name.txt",
    "workspace/data/new_name.txt"
)
# Move a file to a different directory
await sandbox.fs.move_files(
    "workspace/data/file.txt",
    "workspace/archive/file.txt"
)
# Move a directory
await sandbox.fs.move_files(
    "workspace/old_dir",
    "workspace/new_dir"
)

AsyncFileSystem.set_file_permissions

@intercept_errors(message_prefix="Failed to set file permissions: ")
@with_instrumentation()
async def set_file_permissions(path: str,
                               mode: str | None = None,
                               owner: str | None = None,
                               group: str | None = None,
                               request_timeout: float | None = None) -> None

Sets permissions and ownership for a file or directory. Any of the parameters can be None to leave that attribute unchanged.

Arguments:

Example:

# Make a file executable
await sandbox.fs.set_file_permissions(
    path="workspace/scripts/run.sh",
    mode="755"  # rwxr-xr-x
)
# Change file owner
await sandbox.fs.set_file_permissions(
    path="workspace/data/file.txt",
    owner="daytona",
    group="daytona"
)

AsyncFileSystem.upload_file

@overload
async def upload_file(file: bytes,
                      remote_path: str,
                      timeout: int = 30 * 60) -> None

Uploads a file to the specified path in the Sandbox. If a file already exists at the destination path, it will be overwritten. This method is useful when you want to upload small files that fit into memory.

Arguments:

Example:

# Upload a text file
content = b"Hello, World!"
await sandbox.fs.upload_file(content, "tmp/hello.txt")

# Upload a local file
with open("local_file.txt", "rb") as f:
    content = f.read()
await sandbox.fs.upload_file(content, "tmp/file.txt")

# Upload binary data
import json
data = {"key": "value"}
content = json.dumps(data).encode('utf-8')
await sandbox.fs.upload_file(content, "tmp/config.json")

AsyncFileSystem.upload_files

@intercept_errors(message_prefix="Failed to upload files: ")
@with_instrumentation()
async def upload_files(files: list[FileUpload],
                       timeout: int = 30 * 60) -> None

Uploads multiple files to the Sandbox. If files already exist at the destination paths, they will be overwritten.

Arguments:

Example:

# Upload multiple text files
files = [
    FileUpload(
        source=b"Content of file 1",
        destination="/tmp/file1.txt"
    ),
    FileUpload(
        source="workspace/data/file2.txt",
        destination="/tmp/file2.txt"
    ),
    FileUpload(
        source=b'{"key": "value"}',
        destination="/tmp/config.json"
    )
]
await sandbox.fs.upload_files(files)

CancelEvent

@runtime_checkable
class CancelEvent(Protocol)

Duck-typed cancellation token. Compatible with threading.Event and asyncio.Event (both expose is_set()). When supplied to a streaming download, the next chunk read after the event becomes set raises DaytonaError, closing the underlying HTTP connection.

FileUpload

@dataclass
class FileUpload()

Represents a file to be uploaded to the Sandbox.

Attributes:

DownloadProgress

@dataclass
class DownloadProgress()

Progress information for a streaming download.

Attributes:

UploadProgress

@dataclass
class UploadProgress()

Progress information for a streaming upload.

Attributes:

FileDownloadRequest

@dataclass
class FileDownloadRequest()

Represents a request to download a single file from the Sandbox.

Attributes:

FileDownloadResponse

@dataclass
class FileDownloadResponse()

Represents the response to a single file download request.

Attributes: