## AsyncFileSystem

```python
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__

```python
def __init__(api_client: FileSystemApi)
```

Initializes a new FileSystem instance.

**Arguments**:

- `api_client` _FileSystemApi_ \- API client for Sandbox file system operations.

#### AsyncFileSystem.create_folder

```python
@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**:

- `path` _str_ \- Path where the folder should be created. Relative paths are resolved based on the sandbox working directory.
- `mode` _str_ \- Folder permissions in octal format (e.g., “755” for rwxr-xr-x).
- `request_timeout` _float | None_ \- Optional client-side request timeout in seconds. Client-side only. It bounds how long the SDK waits for the HTTP response and does not cancel the operation on the server. Positive values under 1 second are rounded up to 1 second; 0 disables the client-side timeout and negative values are rejected.

**Example**:

```python
# 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

```python
@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**:

- `path` _str_ \- Path to the file to delete. Relative paths are resolved based on the sandbox working directory.
- `recursive` _bool_ \- If the file is a directory, this must be true to delete it.
- `request_timeout` _float | None_ \- Optional client-side request timeout in seconds. Client-side only. It bounds how long the SDK waits for the HTTP response and does not cancel the operation on the server. Positive values under 1 second are rounded up to 1 second; 0 disables the client-side timeout and negative values are rejected.

**Example**:

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

#### AsyncFileSystem.download_file

```python
@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**:

- `remote_path` _str_ \- Path to the file in the Sandbox. Relative paths are resolved based on the sandbox working directory.
- `timeout` _int_ \- Timeout for the download operation in seconds. 0 means no timeout. Default is 30 minutes.

**Returns**:

- `bytes` \- The file contents as a bytes object.

**Example**:

```python
# 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

```python
@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**:

- `remote_path` _str_ \- Path to the file in the Sandbox. Relative paths are resolved based on the sandbox working directory.
- `timeout` _int_ \- Timeout for the download operation in seconds. 0 means no timeout. Default is 30 minutes.
- `on_progress` _Callable[[DownloadProgress], Awaitable[None] | None] | None_ \- Optional callback invoked with cumulative bytes received and total bytes, when known, as the download progresses. May be either a regular function or an `async def` coroutine — coroutine returns are awaited before the next chunk is yielded. Default is None.
- `cancel_event` _CancelEvent | None_ \- Optional `asyncio.Event`-compatible token. When set during streaming, the next chunk raises `DaytonaError` and the underlying HTTP connection is closed. Standard `asyncio.CancelledError` from task cancellation is also honoured automatically by the generator.

**Returns**:

- `AsyncIterator[bytes]` \- An async iterator yielding chunks of file content as bytes.

**Raises**:

- `DaytonaError` \- If the file does not exist, access is denied, or the download is cancelled via `cancel_event`.

**Example**:

```python
# 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

```python
@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**:

- `path` _str_ \- Path to the directory to list contents from. Relative paths are resolved based on the sandbox working directory.
- `depth` _int | None_ \- How many levels deep to list. depth=1 (default) lists the directory’s entries, depth=2 also includes their children, and so on. Must be >= 1. Each returned FileInfo carries a full `path` field.
- `request_timeout` _float | None_ \- Optional client-side request timeout in seconds. Client-side only. It bounds how long the SDK waits for the HTTP response and does not cancel the operation on the server. Positive values under 1 second are rounded up to 1 second; 0 disables the client-side timeout and negative values are rejected.

**Returns**:

- `list[FileInfo]` \- List of file and directory information. Each FileInfo object includes the same fields as described in get_file_info().

**Example**:

```python
# 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

```python
@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**:

- `source` _str_ \- Path to the source file or directory. Relative paths are resolved based on the sandbox working directory.
- `destination` _str_ \- Path to the destination. Relative paths are resolved based on the sandbox working directory.
- `request_timeout` _float | None_ \- Optional client-side request timeout in seconds. Client-side only. It bounds how long the SDK waits for the HTTP response and does not cancel the operation on the server. Positive values under 1 second are rounded up to 1 second; 0 disables the client-side timeout and negative values are rejected.

**Example**:

```python
# 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

```python
@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**:

- `path` _str_ \- Path to the file or directory. Relative paths are resolved based on the sandbox working directory.
- `mode` _str | None_ \- File mode/permissions in octal format (e.g., “644” for rw-r—r—).
- `owner` _str | None_ \- User owner of the file.
- `group` _str | None_ \- Group owner of the file.
- `request_timeout` _float | None_ \- Optional client-side request timeout in seconds. Client-side only. It bounds how long the SDK waits for the HTTP response and does not cancel the operation on the server. Positive values under 1 second are rounded up to 1 second; 0 disables the client-side timeout and negative values are rejected.

**Example**:

```python
# 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

```python
@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**:

- `file` _bytes_ \- File contents as a bytes object.
- `remote_path` _str_ \- Path to the destination file. Relative paths are resolved based on the sandbox working directory.
- `timeout` _int_ \- Timeout for the upload operation in seconds. 0 means no timeout. Default is 30 minutes.

**Example**:

```python
# 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

```python
@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**:

- `files` _list[FileUpload]_ \- List of files to upload.
- `timeout` _int_ \- Timeout for the upload operation in seconds. 0 means no timeout. Default is 30 minutes.

**Example**:

```python
# 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

```python
@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

```python
@dataclass
class FileUpload()
```

Represents a file to be uploaded to the Sandbox.

**Attributes**:

- `source` _bytes | str_ \- File contents as a bytes object or a local file path. If a bytes object is provided, make sure it fits into memory, otherwise use the local file path which content will be streamed to the Sandbox.
- `destination` _str_ \- Absolute destination path in the Sandbox. Relative paths are resolved based on the sandbox working directory.

#### DownloadProgress

```python
@dataclass
class DownloadProgress()
```

Progress information for a streaming download.

**Attributes**:

- `bytes_received` _int_ \- Cumulative bytes received so far.
- `total_bytes` _int | None_ \- Total bytes expected, if known.

#### UploadProgress

```python
@dataclass
class UploadProgress()
```

Progress information for a streaming upload.

**Attributes**:

- `bytes_sent` _int_ \- Cumulative bytes sent so far.

#### FileDownloadRequest

```python
@dataclass
class FileDownloadRequest()
```

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

**Attributes**:

- `source` _str_ \- Source path in the Sandbox. Relative paths are resolved based on the user’s root directory.
- `destination` _str | None_ \- Destination path in the local filesystem where the file content will be streamed to. If not provided, the file will be downloaded in the bytes buffer (might cause memory issues if the file is large).

#### FileDownloadResponse

```python
@dataclass
class FileDownloadResponse()
```

Represents the response to a single file download request.

**Attributes**:

- `source` _str_ \- The original source path requested for download.
- `result` _str | bytes | None_ \- The download result - file path (if destination provided in the request) or bytes content (if no destination in the request), None if failed or no data received.
- `error` _str | None_ \- Error message if the download failed, None if successful.
- `error_details` _FileDownloadErrorDetails | None_ \- Structured error metadata when the server provides it.