AsyncFileSystem

class AsyncFileSystem()

サンドボックス内でのファイルシステム操作を提供します。

このクラスは、Daytona サンドボックス内で実行できるファイルシステム操作に対する高水準インターフェースを実装します。

AsyncFileSystem.init

def __init__(sandbox_id: str, toolbox_api: ToolboxApi,
             get_root_dir: Callable[[], Awaitable[str]])

新しい FileSystem インスタンスを初期化します。

Arguments:

sandbox_id str - サンドボックス ID。
toolbox_api ToolboxApi - サンドボックス操作用の API クライアント。
get_root_dir Callable[[], str] - サンドボックスの既定のルートディレクトリを取得する関数。

AsyncFileSystem.create_folder

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

指定されたパスに、与えられた権限でサンドボックス内に新しいディレクトリを作成します。

Arguments:

path str - フォルダを作成するパス。相対パスはユーザーのルートディレクトリを基準に解決されます。
mode str - 8 進数形式のフォルダ権限（例: “755” は rwxr-xr-x）。

Example:

# 標準的な権限でディレクトリを作成
await sandbox.fs.create_folder("workspace/data", "755")

# プライベートなディレクトリを作成
await sandbox.fs.create_folder("workspace/secrets", "700")

AsyncFileSystem.delete_file

@intercept_errors(message_prefix="Failed to delete file: ")
async def delete_file(path: str) -> None

サンドボックスからファイルを削除します。

Arguments:

path str - 削除するファイルへの絶対パス。

Example:

# ファイルを削除
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

サンドボックスからファイルをダウンロードします。ファイル内容を bytes オブジェクトとして返します。ディスクに保存せずにメモリに読み込みたい場合に便利です。小さなファイルにのみ使用できます。

Arguments:

remote_path str - サンドボックス内のファイルへのパス。相対パスはユーザーのルートディレクトリを基準に解決されます。
timeout int - ダウンロード処理のタイムアウト（秒）。0 はタイムアウトなし。デフォルトは 30 分。

Returns:

bytes - ファイル内容の bytes オブジェクト。

Example:

# ダウンロードしてローカルに保存
content = await sandbox.fs.download_file("workspace/data/file.txt")
with open("local_copy.txt", "wb") as f:
    f.write(content)

# ダウンロードしてテキスト内容を処理
content = await sandbox.fs.download_file("workspace/data/config.json")
config = json.loads(content.decode('utf-8'))

AsyncFileSystem.download_file

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

サンドボックスからファイルをダウンロードし、ストリーミングでローカルファイルに保存します。メモリに収まりきらない可能性のある大きなファイルをダウンロードしたい場合に便利です。

Arguments:

remote_path str - サンドボックス内のファイルへのパス。相対パスはユーザーのルートディレクトリを基準に解決されます。
local_path str - ローカルに保存するファイルのパス。
timeout int - ダウンロード処理のタイムアウト（秒）。0 はタイムアウトなし。デフォルトは 30 分。

Example:

local_path = "local_copy.txt"
await 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")

AsyncFileSystem.find_files

@intercept_errors(message_prefix="Failed to find files: ")
async def find_files(path: str, pattern: str) -> List[Match]

パターンを含むファイルを検索します。grep コマンドに似ています。

引数:

path str - 検索対象のファイルまたはディレクトリへのパス。パスがディレクトリの場合は再帰的に検索されます。相対パスはユーザーのルートディレクトリを基準に解決されます。
pattern str - ファイル内容に対してマッチングを行う検索パターン。

戻り値:

List[Match] - ファイル内で見つかった一致のリスト。各 Match オブジェクトには次が含まれます:
- file: 一致を含むファイルのパス
- line: 一致が見つかった行番号
- content: 一致した行の内容

例:

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

AsyncFileSystem.get_file_info

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

ファイルまたはディレクトリの詳細情報（サイズ、パーミッション、タイムスタンプなど）を取得します。

引数:

path str - ファイルまたはディレクトリへのパス。相対パスはユーザーのルートディレクトリを基準に解決されます。

戻り値:

FileInfo - 次を含む詳細なファイル情報:
- name: ファイル名
- is_dir: パスがディレクトリかどうか
- size: バイト単位のファイルサイズ
- mode: ファイルのパーミッション
- mod_time: 最終更新タイムスタンプ
- permissions: 8進数形式のファイルパーミッション
- owner: ファイルの所有者
- group: ファイルのグループ

例:

# Get file metadata
info = await 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 = await sandbox.fs.get_file_info("workspace/data")
if info.is_dir:
    print("Path is a directory")

AsyncFileSystem.list_files

@intercept_errors(message_prefix="Failed to list files: ")
async def list_files(path: str) -> List[FileInfo]

指定したパス内のファイルとディレクトリを一覧表示し、その情報を返します。ls -l コマンドに似ています。

引数:

path str - 内容を一覧表示するディレクトリへのパス。相対パスはユーザーのルートディレクトリを基準に解決されます。

戻り値:

List[FileInfo] - ファイルおよびディレクトリ情報のリスト。各 FileInfo オブジェクトには、get_file_info() で説明したのと同じフィールドが含まれます。

例:

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

AsyncFileSystem.move_files

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

ファイルまたはディレクトリを移動または名前変更します。移動先の親ディレクトリは存在している必要があります。

引数:

source str - 移動元のファイルまたはディレクトリへのパス。相対パスはユーザーのルートディレクトリを基準に解決されます。
destination str - 移動先のパス。相対パスはユーザーのルートディレクトリを基準に解決されます。

例:

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

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

複数のファイルに対して検索および置換を実行します。

引数:

files List[str] - 置換を実行するファイルパスのリスト。相対パスはユーザーのルートディレクトリを基準に解決されます。
pattern str - 検索するパターン。
new_value str - 一致箇所を置換するテキスト。

戻り値:

List[ReplaceResult] - 各ファイルで行われた置換を示す結果のリスト。各 ReplaceResult には次が含まれます:
- file: 変更されたファイルのパス
- success: 操作が成功したかどうか
- error: 操作が失敗した場合のエラーメッセージ

例:

# 特定のファイルで置換する
results = await sandbox.fs.replace_in_files(
    files=["workspace/src/file1.py", "workspace/src/file2.py"],
    pattern="old_function",
    new_value="new_function"
)

# 結果を出力する
for result in results:
    if result.success:
        print(f"{result.file}: {result.success}")
    else:
        print(f"{result.file}: {result.error}")

AsyncFileSystem.search_files

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

名前が指定のパターンに一致するファイルおよびディレクトリを検索します。パターンは単純な文字列またはグロブパターンに対応します。

引数:

path str - 検索を開始するルートディレクトリへのパス。相対パスはユーザーのルートディレクトリを基準に解決されます。
pattern str - ファイル名と照合するパターン。グロブパターンをサポートします（例: Python ファイルの場合は “*.py”）。

戻り値:

SearchFilesResponse - 次を含む検索結果:
- files: 一致したファイルおよびディレクトリのパス一覧

例:

# すべての Python ファイルを探す
result = await sandbox.fs.search_files("workspace", "*.py")
for file in result.files:
    print(file)

# 特定の接頭辞を持つファイルを探す
result = await sandbox.fs.search_files("workspace/data", "test_*")
print(f"Found {len(result.files)} test files")

AsyncFileSystem.set_file_permissions

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

ファイルまたはディレクトリのパーミッションと所有権を設定します。いずれのパラメータも None にすると、その属性は変更されません。

引数:

path str - ファイルまたはディレクトリへのパス。相対パスはユーザーのルートディレクトリを基準に解決されます。
mode Optional[str] - 8 進数表記のファイルモード／パーミッション（例: “644” は rw-r—r—）。
owner Optional[str] - ファイルの所有ユーザー。
group Optional[str] - ファイルの所有グループ。

例:

# ファイルを実行可能にする
await sandbox.fs.set_file_permissions(
    path="workspace/scripts/run.sh",
    mode="755"  # rwxr-xr-x
)

# ファイルの所有者を変更する
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

サンドボックス内の指定パスにファイルをアップロードします。宛先パスに既にファイルが存在する場合は上書きされます。メモリに収まる小さなファイルをアップロードする際に有用です。

引数:

file bytes - バイト列オブジェクトとしてのファイル内容。
remote_path str - 宛先ファイルへのパス。相対パスはユーザーのルートディレクトリを基準に解決されます。
timeout int - アップロード操作のタイムアウト（秒）。0 はタイムアウトなしを意味します。デフォルトは 30 分です。

例:

# テキストファイルをアップロード
content = b"Hello, World!"
await sandbox.fs.upload_file(content, "tmp/hello.txt")

# ローカルファイルをアップロード
with open("local_file.txt", "rb") as f:
    content = f.read()
await sandbox.fs.upload_file(content, "tmp/file.txt")

# バイナリデータをアップロード
import json
data = {"key": "value"}
content = json.dumps(data).encode('utf-8')
await sandbox.fs.upload_file(content, "tmp/config.json")

AsyncFileSystem.upload_file

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

ローカルファイルシステムからサンドボックス内の指定パスにファイルをアップロードします。宛先パスにすでにファイルが存在する場合は上書きされます。このメソッドはストリーミングでファイルをアップロードするため、メモリに収まりきらない可能性のある大きなファイルのアップロードに有用です。

引数:

local_path str - アップロードするローカルファイルのパス。
remote_path str - サンドボックス内の宛先ファイルのパス。相対パスはユーザーのルートディレクトリを基準に解決されます。
timeout int - アップロード処理のタイムアウト（秒）。0 はタイムアウトなしを意味します。デフォルトは30分。

例:

await sandbox.fs.upload_file("local_file.txt", "tmp/large_file.txt")

AsyncFileSystem.upload_files

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

複数のファイルをサンドボックスにアップロードします。宛先パスにすでにファイルが存在する場合は、上書きされます。

引数:

files List[FileUpload] - アップロードするファイルのリスト。
timeout int - アップロード処理のタイムアウト（秒）。0 はタイムアウトなしを意味します。デフォルトは30分。

例:

# 複数のテキストファイルをアップロード
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)

FileUpload

@dataclass
class FileUpload()

サンドボックスにアップロードするファイルを表します。

属性:

source Union[bytes, str] - バイト列またはローカルファイルパスで指定するファイル内容。バイト列を指定する場合はメモリに収まることを確認し、そうでない場合はローカルファイルパスを使用してください。ローカルファイルの内容はサンドボックスへストリーミングされます。
destination str - サンドボックス内の絶対パス。相対パスはユーザーのルートディレクトリを基準に解決されます。