---
title: "FileSystem"
hideTitleOnPage: true
---

## FileSystem

```python
class FileSystem()
```

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.

#### FileSystem.\_\_init\_\_

```python
def __init__(api_client: FileSystemApi, http_client: httpx.Client)
```

Initializes a new FileSystem instance.

**Arguments**:

- `api_client` _FileSystemApi_ - API client for Sandbox file system operations.
- `http_client` _httpx.Client_ - Shared pooled client for upload/download.

#### FileSystem.create\_folder

```python
@intercept_errors(message_prefix="Failed to create folder: ")
@with_instrumentation()
def create_folder(path: str, mode: str) -> 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).
  

**Example**:

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

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

#### FileSystem.delete\_file

```python
@intercept_errors(message_prefix="Failed to delete file: ")
@with_instrumentation()
def delete_file(path: str, recursive: bool = False) -> 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.
  

**Example**:

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

#### FileSystem.download\_file

```python
@overload
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 = 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 = sandbox.fs.download_file("workspace/data/config.json")
config = json.loads(content.decode('utf-8'))
```

#### FileSystem.download\_file

```python
@overload
def download_file(remote_path: str,
                  local_path: str,
                  timeout: int = 30 * 60) -> None
```

Downloads a file from the Sandbox and saves it to a local file using stream.
This method is useful when you want to download larger files that may not fit into memory.

**Arguments**:

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

**Example**:

```python
local_path = "local_copy.txt"
sandbox.fs.download_file("tmp/large_file.txt", local_path)
size_mb = os.path.getsize(local_path) / 1024 / 1024
print(f"Size of the downloaded file {local_path}: {size_mb} MB")
```

#### FileSystem.download\_file\_stream

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

Downloads a single file from the Sandbox as a stream without buffering the entire file
into memory. Returns an 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], None] | None_ - Optional callback invoked with
  cumulative bytes received and total bytes, when known, as the download progresses.
  Default is None.
- `cancel_event` _CancelEvent | None_ - Optional ``threading.Event``-compatible token. When
  set during streaming, the next chunk raises ``DaytonaError`` and the underlying
  HTTP connection is closed.
  

**Returns**:

- `Iterator[bytes]` - An 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
with open("local_copy.bin", "wb") as f:
    for chunk in sandbox.fs.download_file_stream("workspace/large-file.bin"):
        f.write(chunk)

# Cancel an in-progress download from another thread
import threading
cancel = threading.Event()
threading.Timer(5.0, cancel.set).start()
for chunk in sandbox.fs.download_file_stream("workspace/big.bin", cancel_event=cancel):
    process(chunk)
```

#### FileSystem.download\_files

```python
@intercept_errors(message_prefix="Failed to download files: ")
@with_instrumentation()
def download_files(files: list[FileDownloadRequest],
                   timeout: int = 30 * 60) -> list[FileDownloadResponse]
```

Downloads multiple files from the Sandbox. If the files already exist locally, they will be overwritten.

**Arguments**:

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

**Returns**:

- `list[FileDownloadResponse]` - List of download results.
  

**Raises**:

- `Exception` - Only if the request itself fails (network issues, invalid request/response, etc.). Individual
  file download errors are returned in `FileDownloadResponse.error`. When the daemon provides structured
  per-file metadata, it is also available in `FileDownloadResponse.error_details`.
  

**Example**:

```python
# Download multiple files
results = sandbox.fs.download_files([
    FileDownloadRequest(source="tmp/data.json"),
    FileDownloadRequest(source="tmp/config.json", destination="local_config.json")
])
for result in results:
    if result.error:
        print(f"Error downloading {result.source}: {result.error}")
    elif result.result:
        print(f"Downloaded {result.source} to {result.result}")
```

#### FileSystem.find\_files

```python
@intercept_errors(message_prefix="Failed to find files: ")
@with_instrumentation()
def find_files(path: str, pattern: str) -> list[Match]
```

Searches for files containing a pattern, similar to
the grep command.

**Arguments**:

- `path` _str_ - Path to the file or directory to search. If the path is a directory,
  the search will be performed recursively. Relative paths are resolved based
  on the sandbox working directory.
- `pattern` _str_ - Search pattern to match against file contents.
  

**Returns**:

- `list[Match]` - List of matches found in files. Each Match object includes:
  - file: Path to the file containing the match
  - line: The line number where the match was found
  - content: The matching line content
  

**Example**:

```python
# Search for TODOs in Python files
matches = sandbox.fs.find_files("workspace/src", "TODO:")
for match in matches:
    print(f"{match.file}:{match.line}: {match.content.strip()}")
```

#### FileSystem.get\_file\_info

```python
@intercept_errors(message_prefix="Failed to get file info: ")
@with_instrumentation()
def get_file_info(path: str) -> FileInfo
```

Gets detailed information about a file or directory, including its
size, permissions, and timestamps.

**Arguments**:

- `path` _str_ - Path to the file or directory. Relative paths are resolved based
  on the sandbox working directory.
  

**Returns**:

- `FileInfo` - Detailed file information including:
  - name: File name
  - is_dir: Whether the path is a directory
  - size: File size in bytes
  - mode: File permissions
  - mod_time: Last modification timestamp
  - permissions: File permissions in octal format
  - owner: File owner
  - group: File group
  

**Example**:

```python
# Get file metadata
info = sandbox.fs.get_file_info("workspace/data/file.txt")
print(f"Size: {info.size} bytes")
print(f"Modified: {info.mod_time}")
print(f"Mode: {info.mode}")

# Check if path is a directory
info = sandbox.fs.get_file_info("workspace/data")
if info.is_dir:
    print("Path is a directory")
```

#### FileSystem.list\_files

```python
@intercept_errors(message_prefix="Failed to list files: ")
@with_instrumentation()
def list_files(path: str) -> 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.
  

**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 = 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 only directories
dirs = [f for f in files if f.is_dir]
print("Subdirectories:", ", ".join(d.name for d in dirs))
```

#### FileSystem.move\_files

```python
@intercept_errors(message_prefix="Failed to move files: ")
@with_instrumentation()
def move_files(source: str, destination: str) -> 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.
  

**Example**:

```python
# Rename a file
sandbox.fs.move_files(
    "workspace/data/old_name.txt",
    "workspace/data/new_name.txt"
)

# Move a file to a different directory
sandbox.fs.move_files(
    "workspace/data/file.txt",
    "workspace/archive/file.txt"
)

# Move a directory
sandbox.fs.move_files(
    "workspace/old_dir",
    "workspace/new_dir"
)
```

#### FileSystem.replace\_in\_files

```python
@intercept_errors(message_prefix="Failed to replace in files: ")
@with_instrumentation()
def replace_in_files(files: list[str], pattern: str,
                     new_value: str) -> list[ReplaceResult]
```

Performs search and replace operations across multiple files.

**Arguments**:

- `files` _list[str]_ - List of file paths to perform replacements in. Relative paths are
  resolved based on the sandbox working directory.
- `pattern` _str_ - Pattern to search for.
- `new_value` _str_ - Text to replace matches with.
  

**Returns**:

- `list[ReplaceResult]` - List of results indicating replacements made in
  each file. Each ReplaceResult includes:
  - file: Path to the modified file
  - success: Whether the operation was successful
  - error: Error message if the operation failed
  

**Example**:

```python
# Replace in specific files
results = sandbox.fs.replace_in_files(
    files=["workspace/src/file1.py", "workspace/src/file2.py"],
    pattern="old_function",
    new_value="new_function"
)

# Print results
for result in results:
    if result.success:
        print(f"{result.file}: {result.success}")
    else:
        print(f"{result.file}: {result.error}")
```

#### FileSystem.search\_files

```python
@intercept_errors(message_prefix="Failed to search files: ")
@with_instrumentation()
def search_files(path: str, pattern: str) -> SearchFilesResponse
```

Searches for files and directories whose names match the
specified pattern. The pattern can be a simple string or a glob pattern.

**Arguments**:

- `path` _str_ - Path to the root directory to start search from. Relative paths are resolved
  based on the sandbox working directory.
- `pattern` _str_ - Pattern to match against file names. Supports glob
  patterns (e.g., "*.py" for Python files).
  

**Returns**:

- `SearchFilesResponse` - Search results containing:
  - files: List of matching file and directory paths
  

**Example**:

```python
# Find all Python files
result = sandbox.fs.search_files("workspace", "*.py")
for file in result.files:
    print(file)

# Find files with specific prefix
result = sandbox.fs.search_files("workspace/data", "test_*")
print(f"Found {len(result.files)} test files")
```

#### FileSystem.set\_file\_permissions

```python
@intercept_errors(message_prefix="Failed to set file permissions: ")
@with_instrumentation()
def set_file_permissions(path: str,
                         mode: str | None = None,
                         owner: str | None = None,
                         group: str | 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.
  

**Example**:

```python
# Make a file executable
sandbox.fs.set_file_permissions(
    path="workspace/scripts/run.sh",
    mode="755"  # rwxr-xr-x
)

# Change file owner
sandbox.fs.set_file_permissions(
    path="workspace/data/file.txt",
    owner="daytona",
    group="daytona"
)
```

#### FileSystem.upload\_file

```python
@overload
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!"
sandbox.fs.upload_file(content, "tmp/hello.txt")

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

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

#### FileSystem.upload\_file

```python
@overload
def upload_file(local_path: str,
                remote_path: str,
                timeout: int = 30 * 60) -> None
```

Uploads a file from the local file system to the specified path in the Sandbox.
If a file already exists at the destination path, it will be overwritten. This method uses
streaming to upload the file, so it is useful when you want to upload larger files that may
not fit into memory.

**Arguments**:

- `local_path` _str_ - Path to the local file to upload.
- `remote_path` _str_ - Path to the destination file in the Sandbox. 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
sandbox.fs.upload_file("local_file.txt", "tmp/large_file.txt")
```

#### FileSystem.upload\_files

```python
@intercept_errors(message_prefix="Failed to upload files: ")
@with_instrumentation()
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"
    )
]
sandbox.fs.upload_files(files)
```

#### FileSystem.upload\_file\_stream

```python
@intercept_errors(message_prefix="Failed to upload file: ")
@with_instrumentation()
def upload_file_stream(source: bytes | bytearray | str | io.IOBase,
                       remote_path: str,
                       timeout: int = 30 * 60,
                       on_progress: Callable[[UploadProgress], None]
                       | None = None,
                       cancel_event: CancelEvent | None = None) -> None
```

Uploads a single file to the Sandbox using true streaming, with optional progress
tracking and cancellation. Memory usage stays flat regardless of source size. The
HTTP layer uses chunked transfer encoding, so the source's natural EOF terminates
the upload — no advance size is needed.

**Arguments**:

- `source` _bytes | bytearray | str | IOBase_ - Data to upload. ``bytes`` is uploaded
  from memory; ``str`` is treated as a local file path and read in chunks; a
  file-like object (anything implementing ``.read()``) is streamed as-is.
- `remote_path` _str_ - Destination path in the Sandbox.
- `timeout` _int_ - Timeout in seconds. 0 means no timeout. Default is 30 minutes.
- `on_progress` _Callable[[UploadProgress], None] | None_ - Optional callback invoked
  with cumulative bytes sent.
- `cancel_event` _CancelEvent | None_ - Optional ``threading.Event``-compatible token.
  When set during streaming, the next chunk read raises ``DaytonaError`` and
  tears down the request.
  

**Raises**:

- `DaytonaError` - If the upload fails or is cancelled via ``cancel_event``.
  

**Example**:

```python
import threading
cancel = threading.Event()
with open("large_dataset.csv", "rb") as f:
    sandbox.fs.upload_file_stream(
        f,
        "tmp/dataset.csv",
        on_progress=lambda p: print(f"{p.bytes_sent} bytes sent"),
        cancel_event=cancel,
    )
```


## 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.

## FileDownloadErrorDetails

```python
@dataclass
class FileDownloadErrorDetails()
```

Structured error metadata for a failed bulk file download item.

#### create\_file\_download\_error

```python
def create_file_download_error(response: FileDownloadResponse) -> DaytonaError
```

Create the appropriate Daytona exception for a failed file download response.

#### raise\_if\_stream\_error

```python
def raise_if_stream_error(remote_path: str, error_text: str | None,
                          error_details: FileDownloadErrorDetails | None,
                          received_file_data: bool) -> None
```

Raise the appropriate error after streaming a single-file multipart download.

#### parse\_file\_download\_error\_payload

```python
def parse_file_download_error_payload(
        payload: bytes, content_type: str | None
) -> tuple[str, FileDownloadErrorDetails | None]
```

Parse a bulk-download error part into the legacy message and structured metadata.

