FileInfo
class FileInfo(BaseModel)One filesystem entry as /files and /files/info serialize it.
Attributes:
groupStrictStr = Field(description='Numeric group id rendered as a string.')is_dirStrictBool = Field(description='Whether the entry is a directory.', alias='isDir')mod_timeStrictStr = Field(description='Deprecated but load-bearing: Gotime.String()layout for the last modification time.', alias='modTime')modeStrictStr = Field(description='Goos.FileMode.String()layout (-rw-r--r--,drwxr-xr-x, …).')modified_atStrictStr = Field(description='RFC 3339 UTC timestamp for the last modification time.', alias='modifiedAt')nameStrictStr = Field(description='Base name of the entry.')ownerStrictStr = Field(description='Numeric owner id rendered as a string.')pathStrictStr = Field(description="Absolute path to the entry, rendered in the guest's path syntax.")permissionsStrictStr = Field(description='Four-digit octal permissions (0644,0755, …).')sizeAnnotated[int, Field(strict=True, ge=0)] = Field(description='Entry size in bytes.')additional_propertiesDict[str, Any] = {}
model_config
model_config = ConfigDict(populate_by_name=True, validate_assignment=True, protected_namespaces=())FileInfo.to_str
def to_str() -> strReturns the string representation of the model using alias
FileInfo.to_json
def to_json() -> strReturns 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):
Noneis only added to the output dict for nullable fields that were set at model initialization. Other fields with valueNoneare ignored.- Fields in
self.additional_propertiesare 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
WatchHandle
class WatchHandle()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.
WatchHandle.__init__
def __init__(stream: _LineStream,
events: asyncio.Queue[FilesystemEvent],
on_exit: Optional[_ExitCallback] = None) -> NoneWatchHandle.overflow_count
@property
def overflow_count() -> intWatchHandle.attach_reader
def attach_reader(reader: asyncio.Task[None]) -> NoneWatchHandle.exit_error
@property
def exit_error() -> Exception | NoneWatchHandle.last_callback_error
@property
def last_callback_error() -> Exception | NoneWatchHandle.record_callback_error
def record_callback_error(error: Exception) -> NoneWatchHandle.record_exit
async def record_exit(error: Exception | None) -> NoneWatchHandle.record_overflow
def record_overflow() -> NoneWatchHandle.wait_for_exit
async def wait_for_exit() -> Exception | NoneWatchHandle.get_new_events
def get_new_events() -> list[FilesystemEvent]WatchHandle.stop
async def stop() -> NoneFileSystemService
class FileSystemService()Sandbox filesystem namespace.
FileSystemService.__init__
def __init__(transport: AsyncTransport,
*,
file_url_signer: Optional[FileUrlSigner] = None) -> NoneFileSystemService.read_bytes
async def read_bytes(path: str) -> bytesRead a file as bytes.
FileSystemService.read_text
async def read_text(path: str) -> strRead a UTF-8 text file.
FileSystemService.write_bytes
async def write_bytes(path: str, content: bytes, *, atomic: bool = False) -> NoneWrite bytes to a file, optionally via temp-file + rename.
FileSystemService.write_text
async def write_text(path: str, content: str, *, atomic: bool = False) -> NoneWrite a UTF-8 text file, optionally via temp-file + rename.
FileSystemService.presigned_download_url
async def presigned_download_url(path: str, *, expires_in_seconds: int = 3600) -> strReturn 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).
FileSystemService.presigned_upload_url
async def presigned_upload_url(path: str, *, expires_in_seconds: int = 3600) -> strReturn 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).
FileSystemService.write_many
async def write_many(files: Mapping[str, bytes],
*,
atomic: bool = False) -> BulkWriteResultWrite several files in one request, optionally atomically per file.
FileSystemService.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) -> NoneStream a local file into the sandbox without buffering it whole in memory.
FileSystemService.download_file
async def download_file(remote_path: str,
local_path: str | os.PathLike[str],
*,
on_progress: TransferProgressCallback | None = None,
cancel: TransferCancelSignal | None = None) -> NoneFileSystemService.read_many
async def read_many(paths: Sequence[str]) -> BulkReadResultRead 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.
FileSystemService.list
async def list(path: str, *, depth: Optional[int] = None) -> ListResult[FileInfo]List directory entries as typed FileInfo rows (items, next_cursor).
FileSystemService.stat
async def stat(path: str) -> FileInfoReturn file metadata.
FileSystemService.mkdir
async def mkdir(path: str, *, mode: str = '755') -> NoneCreate a directory.
FileSystemService.delete
async def delete(path: str, *, recursive: bool = False) -> NoneDelete a file or directory.
FileSystemService.move
async def move(source: str, destination: str) -> NoneMove or rename files.
FileSystemService.find
async def find(path: str, pattern: str) -> ListResultFind files under a path.
FileSystemService.search
async def search(path: str, pattern: str) -> ListResult[str]Search file contents.
FileSystemService.replace
async def replace(files: Sequence[str], pattern: str, new_value: str) -> ListResultReplace text in files.
FileSystemService.chmod
async def chmod(path: str,
*,
mode: Optional[str] = None,
owner: Optional[str] = None,
group: Optional[str] = None) -> NoneSet file permissions and ownership.
FileSystemService.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) -> WatchHandleTransferCancelSignal
class TransferCancelSignal(Protocol)A cancellation signal for file transfers.
asyncio.Event and threading.Event both satisfy this protocol.
TransferCancelSignal.is_set
def is_set() -> boolFilesystemEventType
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). partial_regions
names the region slugs whose cell has not finished backfilling the index
(or has never heartbeated); results for them may be incomplete. Empty in
steady state.
Attributes:
itemsList[T]next_cursorOptional[str] = Nonetotalint = 0total_relationstr = 'eq'partial_regionsList[str] = field(default_factory=list)
FilesystemEvent
@dataclass(frozen=True, slots=True)
class FilesystemEvent()Attributes:
typeFilesystemEventTypenamestris_dirbooltimestampint
BulkWriteFile
@dataclass(frozen=True)
class BulkWriteFile()Attributes:
pathstrbytesint
BulkWriteResult
@dataclass(frozen=True)
class BulkWriteResult()Attributes:
filesList[BulkWriteFile]errorsList[str] = field(default_factory=list)
BulkReadError
@dataclass(frozen=True)
class BulkReadError()Attributes:
pathstrmessagestrcodeOptional[str] = NonesourceOptional[str] = None
BulkReadResult
@dataclass(frozen=True)
class BulkReadResult()Attributes:
filesDict[str, bytes]errorsList[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_prefixOptional[str] = Noneid_prefixOptional[str] = NonecursorOptional[str] = NonelimitOptional[int] = None
SyncFileSystemService
class SyncFileSystemService()Sandbox filesystem namespace.
SyncFileSystemService.__init__
def __init__(impl: FileSystemService, bridge: SyncBridge) -> NoneSyncFileSystemService.chmod
def chmod(path: str,
*,
mode: Optional[str] = None,
owner: Optional[str] = None,
group: Optional[str] = None) -> NoneSet file permissions and ownership.
SyncFileSystemService.delete
def delete(path: str, *, recursive: bool = False) -> NoneDelete a file or directory.
SyncFileSystemService.download_file
def download_file(remote_path: str,
local_path: str | os.PathLike[str],
*,
on_progress: TransferProgressCallback | None = None,
cancel: TransferCancelSignal | None = None) -> NoneSyncFileSystemService.find
def find(path: str, pattern: str) -> ListResultFind files under a path.
SyncFileSystemService.list
def list(path: str, *, depth: Optional[int] = None) -> ListResult[FileInfo]List directory entries as typed FileInfo rows (items, next_cursor).
SyncFileSystemService.mkdir
def mkdir(path: str, *, mode: str = '755') -> NoneCreate a directory.
SyncFileSystemService.move
def move(source: str, destination: str) -> NoneMove or rename files.
SyncFileSystemService.presigned_download_url
def presigned_download_url(path: str, *, expires_in_seconds: int = 3600) -> strReturn a pre-signed download URL for a sandbox file.
SyncFileSystemService.presigned_upload_url
def presigned_upload_url(path: str, *, expires_in_seconds: int = 3600) -> strReturn a pre-signed upload URL for a sandbox file.
SyncFileSystemService.read_bytes
def read_bytes(path: str) -> bytesRead a file as bytes.
SyncFileSystemService.read_many
def read_many(paths: Sequence[str]) -> BulkReadResultRead multiple files, buffering each returned file in memory before returning.
SyncFileSystemService.read_text
def read_text(path: str) -> strRead a UTF-8 text file.
SyncFileSystemService.replace
def replace(files: Sequence[str], pattern: str, new_value: str) -> ListResultReplace text in files.
SyncFileSystemService.search
def search(path: str, pattern: str) -> ListResult[str]Search file contents.
SyncFileSystemService.stat
def stat(path: str) -> FileInfoReturn file metadata.
SyncFileSystemService.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) -> NoneStream a local file into the sandbox without buffering it whole in memory.
SyncFileSystemService.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) -> SyncWatchHandleSyncFileSystemService.write_bytes
def write_bytes(path: str, content: bytes, *, atomic: bool = False) -> NoneWrite bytes to a file, optionally via temp-file + rename.
SyncFileSystemService.write_many
def write_many(files: Mapping[str, bytes], *, atomic: bool = False) -> BulkWriteResultWrite several files in one request, optionally atomically per file.
SyncFileSystemService.write_text
def write_text(path: str, content: str, *, atomic: bool = False) -> NoneWrite a UTF-8 text file, optionally via temp-file + rename.
SyncWatchHandle
class SyncWatchHandle()A running directory watch.
SyncWatchHandle.__init__
def __init__(impl: WatchHandle, bridge: SyncBridge) -> NoneSyncWatchHandle.attach_reader
def attach_reader(reader: asyncio.Task[None]) -> NoneSyncWatchHandle.get_new_events
def get_new_events() -> list[FilesystemEvent]SyncWatchHandle.record_callback_error
def record_callback_error(error: Exception) -> NoneSyncWatchHandle.record_exit
def record_exit(error: Exception | None) -> NoneSyncWatchHandle.record_overflow
def record_overflow() -> NoneSyncWatchHandle.stop
def stop() -> NoneSyncWatchHandle.wait_for_exit
def wait_for_exit() -> Exception | NoneSyncWatchHandle.exit_error
@property
def exit_error() -> Exception | NoneSyncWatchHandle.last_callback_error
@property
def last_callback_error() -> Exception | NoneSyncWatchHandle.overflow_count
@property
def overflow_count() -> intSIGNED_FILE_MAX_TTL_SECONDS
SIGNED_FILE_MAX_TTL_SECONDS = 604800TransferProgressCallback
TransferProgressCallback = Callable[[int, Optional[int]], None]SearchSort
SearchSort = Literal['created_at', 'updated_at', 'last_activity_at', 'name', 'vcpus', 'mem_mib', 'scratch_mib']SearchOrder
SearchOrder = Literal['asc', 'desc']SearchTimestamp
SearchTimestamp = Union[str, datetime]An RFC 3339 string, or a datetime (naive values are taken as UTC).