Daytona v2

Documentation for Daytona v2, currently in alpha. For the latest release, see Daytona v0.

FileSystemService

class FileSystemService()

Sandbox filesystem namespace.

FileSystemService.__init__

def __init__(impl: AsyncFileSystemService, bridge: SyncBridge) -> None

FileSystemService.chmod

def chmod(path: str,
          *,
          mode: Optional[str] = None,
          owner: Optional[str] = None,
          group: Optional[str] = None) -> None

Set file permissions and ownership.

FileSystemService.delete

def delete(path: str, *, recursive: bool = False) -> None

Delete a file or directory.

FileSystemService.download_file

def download_file(remote_path: str,
                  local_path: str | os.PathLike[str],
                  *,
                  on_progress: TransferProgressCallback | None = None,
                  cancel: TransferCancelSignal | None = None) -> None

FileSystemService.find

def find(path: str, pattern: str) -> ListResult

Find files under a path.

FileSystemService.list

def list(path: str, *, depth: Optional[int] = None) -> ListResult[FileInfo]

List directory entries as typed FileInfo rows (items, next_cursor).

FileSystemService.mkdir

def mkdir(path: str, *, mode: str = '755') -> None

Create a directory.

FileSystemService.move

def move(source: str, destination: str) -> None

Move or rename files.

FileSystemService.presigned_download_url

def presigned_download_url(path: str, *, expires_in_seconds: int = 3600) -> str

Return a pre-signed download URL for a sandbox file.

FileSystemService.presigned_upload_url

def presigned_upload_url(path: str, *, expires_in_seconds: int = 3600) -> str

Return a pre-signed upload URL for a sandbox file.

FileSystemService.read_bytes

def read_bytes(path: str) -> bytes

Read a file as bytes.

FileSystemService.read_many

def read_many(paths: Sequence[str]) -> BulkReadResult

Read multiple files, buffering each returned file in memory before returning.

FileSystemService.read_text

def read_text(path: str) -> str

Read a UTF-8 text file.

FileSystemService.replace

def replace(files: Sequence[str], pattern: str, new_value: str) -> ListResult

Replace text in files.

FileSystemService.search

def search(path: str, pattern: str) -> ListResult[str]

Search file contents.

FileSystemService.stat

def stat(path: str) -> FileInfo

Return file metadata.

FileSystemService.upload_file

def upload_file(local_path: str | os.PathLike[str],
                remote_path: str,
                *,
                on_progress: TransferProgressCallback | None = None,
                cancel: TransferCancelSignal | None = None,
                atomic: bool = False) -> None

Stream a local file into the sandbox without buffering it whole in memory.

FileSystemService.watch_dir

def watch_dir(path: str,
              on_event: Optional[_WatchCallback] = None,
              *,
              recursive: bool = False,
              on_overflow: Optional[_OverflowCallback] = None,
              on_exit: Optional[_ExitCallback] = None) -> WatchHandle

FileSystemService.write_bytes

def write_bytes(path: str, content: bytes, *, atomic: bool = False) -> None

Write bytes to a file, optionally via temp-file + rename.

FileSystemService.write_many

def write_many(files: Mapping[str, bytes], *, atomic: bool = False) -> BulkWriteResult

Write several files in one request, optionally atomically per file.

FileSystemService.write_text

def write_text(path: str, content: str, *, atomic: bool = False) -> None

Write a UTF-8 text file, optionally via temp-file + rename.

AsyncFileSystemService

class AsyncFileSystemService()

Sandbox filesystem namespace.

AsyncFileSystemService.__init__

def __init__(transport: AsyncTransport,
             *,
             file_url_signer: Optional[FileUrlSigner] = None) -> None

AsyncFileSystemService.read_bytes

async def read_bytes(path: str) -> bytes

Read a file as bytes.

AsyncFileSystemService.read_text

async def read_text(path: str) -> str

Read a UTF-8 text file.

AsyncFileSystemService.write_bytes

async def write_bytes(path: str, content: bytes, *, atomic: bool = False) -> None

Write bytes to a file, optionally via temp-file + rename.

AsyncFileSystemService.write_text

async def write_text(path: str, content: str, *, atomic: bool = False) -> None

Write a UTF-8 text file, optionally via temp-file + rename.

AsyncFileSystemService.presigned_download_url

async def presigned_download_url(path: str, *, expires_in_seconds: int = 3600) -> str

Return a pre-signed download URL for a sandbox file.

The signature is minted by the control plane; expires_in_seconds must be between 1 and SIGNED_FILE_MAX_TTL_SECONDS (7 days).

AsyncFileSystemService.presigned_upload_url

async def presigned_upload_url(path: str, *, expires_in_seconds: int = 3600) -> str

Return a pre-signed upload URL for a sandbox file.

The signature is minted by the control plane; expires_in_seconds must be between 1 and SIGNED_FILE_MAX_TTL_SECONDS (7 days).

AsyncFileSystemService.write_many

async def write_many(files: Mapping[str, bytes],
                     *,
                     atomic: bool = False) -> BulkWriteResult

Write several files in one request, optionally atomically per file.

AsyncFileSystemService.upload_file

async def upload_file(local_path: str | os.PathLike[str],
                      remote_path: str,
                      *,
                      on_progress: TransferProgressCallback | None = None,
                      cancel: TransferCancelSignal | None = None,
                      atomic: bool = False) -> None

Stream a local file into the sandbox without buffering it whole in memory.

AsyncFileSystemService.download_file

async def download_file(remote_path: str,
                        local_path: str | os.PathLike[str],
                        *,
                        on_progress: TransferProgressCallback | None = None,
                        cancel: TransferCancelSignal | None = None) -> None

AsyncFileSystemService.read_many

async def read_many(paths: Sequence[str]) -> BulkReadResult

Read multiple files, buffering each returned file in memory before returning.

Non-200 responses raise the typed SDK error inside the transport; only a 200 multipart body reaches the parser.

AsyncFileSystemService.list

async def list(path: str, *, depth: Optional[int] = None) -> ListResult[FileInfo]

List directory entries as typed FileInfo rows (items, next_cursor).

AsyncFileSystemService.stat

async def stat(path: str) -> FileInfo

Return file metadata.

AsyncFileSystemService.mkdir

async def mkdir(path: str, *, mode: str = '755') -> None

Create a directory.

AsyncFileSystemService.delete

async def delete(path: str, *, recursive: bool = False) -> None

Delete a file or directory.

AsyncFileSystemService.move

async def move(source: str, destination: str) -> None

Move or rename files.

AsyncFileSystemService.find

async def find(path: str, pattern: str) -> ListResult

Find files under a path.

AsyncFileSystemService.search

async def search(path: str, pattern: str) -> ListResult[str]

Search file contents.

AsyncFileSystemService.replace

async def replace(files: Sequence[str], pattern: str, new_value: str) -> ListResult

Replace text in files.

AsyncFileSystemService.chmod

async def chmod(path: str,
                *,
                mode: Optional[str] = None,
                owner: Optional[str] = None,
                group: Optional[str] = None) -> None

Set file permissions and ownership.

AsyncFileSystemService.watch_dir

async def watch_dir(path: str,
                    on_event: Optional[_WatchCallback] = None,
                    *,
                    recursive: bool = False,
                    on_overflow: Optional[_OverflowCallback] = None,
                    on_exit: Optional[_ExitCallback] = None) -> AsyncWatchHandle

WatchHandle

class WatchHandle()

A running directory watch.

WatchHandle.__init__

def __init__(impl: AsyncWatchHandle, bridge: SyncBridge) -> None

WatchHandle.attach_reader

def attach_reader(reader: asyncio.Task[None]) -> None

WatchHandle.get_new_events

def get_new_events() -> list[FilesystemEvent]

WatchHandle.record_callback_error

def record_callback_error(error: Exception) -> None

WatchHandle.record_exit

def record_exit(error: Exception | None) -> None

WatchHandle.record_overflow

def record_overflow() -> None

WatchHandle.stop

def stop() -> None

WatchHandle.wait_for_exit

def wait_for_exit() -> Exception | None

WatchHandle.exit_error

@property
def exit_error() -> Exception | None

WatchHandle.last_callback_error

@property
def last_callback_error() -> Exception | None

WatchHandle.overflow_count

@property
def overflow_count() -> int

AsyncWatchHandle

class AsyncWatchHandle()

A running directory watch.

Single-task, single-loop: the reader is one asyncio.Task and every method runs on the same loop, so the sync version’s five locks collapse to plain attributes. Only the exit event survives, because wait_for_exit genuinely waits.

AsyncWatchHandle.__init__

def __init__(stream: _LineStream,
             events: asyncio.Queue[FilesystemEvent],
             on_exit: Optional[_ExitCallback] = None) -> None

AsyncWatchHandle.overflow_count

@property
def overflow_count() -> int

AsyncWatchHandle.attach_reader

def attach_reader(reader: asyncio.Task[None]) -> None

AsyncWatchHandle.exit_error

@property
def exit_error() -> Exception | None

AsyncWatchHandle.last_callback_error

@property
def last_callback_error() -> Exception | None

AsyncWatchHandle.record_callback_error

def record_callback_error(error: Exception) -> None

AsyncWatchHandle.record_exit

async def record_exit(error: Exception | None) -> None

AsyncWatchHandle.record_overflow

def record_overflow() -> None

AsyncWatchHandle.wait_for_exit

async def wait_for_exit() -> Exception | None

AsyncWatchHandle.get_new_events

def get_new_events() -> list[FilesystemEvent]

AsyncWatchHandle.stop

async def stop() -> None

FileInfo

class FileInfo(BaseModel)

One filesystem entry as /files and /files/info serialize it.

Attributes:

• group StrictStr = Field(description=‘Numeric group id rendered as a string.’)
• is_dir StrictBool = Field(description=‘Whether the entry is a directory.’, alias=‘isDir’)
• mod_time StrictStr = Field(description=‘Deprecated but load-bearing: Go time.String() layout for the last modification time.’, alias=‘modTime’)
• mode StrictStr = Field(description=‘Go os.FileMode.String() layout (-rw-r--r--, drwxr-xr-x, …).’)
• modified_at StrictStr = Field(description=‘RFC 3339 UTC timestamp for the last modification time.’, alias=‘modifiedAt’)
• name StrictStr = Field(description=‘Base name of the entry.’)
• owner StrictStr = Field(description=‘Numeric owner id rendered as a string.’)
• path StrictStr = Field(description=“Absolute path to the entry, rendered in the guest’s path syntax.”)
• permissions StrictStr = Field(description=‘Four-digit octal permissions (0644, 0755, …).’)
• size Annotated[int, Field(strict=True, ge=0)] = Field(description=‘Entry size in bytes.’)
• additional_properties Dict[str, Any] = {}

model_config

model_config = ConfigDict(populate_by_name=True, validate_assignment=True, protected_namespaces=())

FileInfo.to_str

def to_str() -> str

Returns the string representation of the model using alias

FileInfo.to_json

def to_json() -> str

Returns the JSON representation of the model using alias

FileInfo.from_json

@classmethod
def from_json(json_str: str) -> Optional[Self]

Create an instance of FileInfo from a JSON string

FileInfo.to_dict

def to_dict() -> Dict[str, Any]

Return the dictionary representation of the model using alias.

This has the following differences from calling pydantic’s self.model_dump(by_alias=True):

• None is only added to the output dict for nullable fields that were set at model initialization. Other fields with value None are ignored.
• Fields in self.additional_properties are added to the output dict.

FileInfo.from_dict

@classmethod
def from_dict(obj: Optional[Dict[str, Any]]) -> Optional[Self]

Create an instance of FileInfo from a dict

TransferCancelSignal

class TransferCancelSignal(Protocol)

A cancellation signal for file transfers.

asyncio.Event and threading.Event both satisfy this protocol.

TransferCancelSignal.is_set

def is_set() -> bool

FilesystemEventType

class FilesystemEventType(str, Enum)

Attributes:

• CREATE ‘create’
• WRITE ‘write’
• REMOVE ‘remove’
• RENAME ‘rename’
• CHMOD ‘chmod’

SearchResult

@dataclass(frozen=True)
class SearchResult(Generic[T])

One page of GET /sandboxes/search (Daytona.search()).

total is exact when total_relation == "eq" and a lower bound when it is "gte" (the index caps exact counting at 10 000).

Attributes:

• items List[T]
• next_cursor Optional[str] = None
• total int = 0
• total_relation SearchTotalRelation = ‘eq’
• raw_total_relation str = ‘eq’

FilesystemEvent

@dataclass(frozen=True, slots=True)
class FilesystemEvent()

Attributes:

• type FilesystemEventType
• name str
• is_dir bool
• timestamp int

BulkWriteFile

@dataclass(frozen=True)
class BulkWriteFile()

Attributes:

• path str
• bytes int

BulkWriteResult

@dataclass(frozen=True)
class BulkWriteResult()

Attributes:

• files List[BulkWriteFile]
• errors List[str] = field(default_factory=list)

BulkReadError

@dataclass(frozen=True)
class BulkReadError()

Attributes:

• path str
• message str
• code Optional[str] = None
• source Optional[str] = None

BulkReadResult

@dataclass(frozen=True)
class BulkReadResult()

Attributes:

• files Dict[str, bytes]
• errors List[BulkReadError] = field(default_factory=list)

SearchSandboxesQuery

@dataclass(frozen=True, kw_only=True)
class SearchSandboxesQuery(SandboxSearchFilters)

Every Daytona.search() parameter (GET /sandboxes/search); keyword-only.

Attributes:

• name_prefix Optional[str] = None
• id_prefix Optional[str] = None
• cursor Optional[str] = None
• limit Optional[int] = None

SIGNED_FILE_MAX_TTL_SECONDS

SIGNED_FILE_MAX_TTL_SECONDS = 604800

TransferProgressCallback

TransferProgressCallback = Callable[[int, Optional[int]], None]

SearchTotalRelation

SearchTotalRelation = Literal['eq', 'gte', 'unknown']

SearchSort

SearchSort = Literal['created_at', 'updated_at', 'last_activity_at', 'name', 'vcpus', 'memory_mib', 'disk_mib']

SearchOrder

SearchOrder = Literal['asc', 'desc']

SearchTimestamp

SearchTimestamp = Union[str, datetime]

An RFC 3339 string, or a datetime (naive values are taken as UTC).