files.watchDir() メソッドを、Python では files.watch_dir() メソッドを使用して、ディレクトリの変更を監視できます。
イベントは非同期で追跡されるため、配信が遅れる場合があります。
変更を行った直後に watcher を収集したり閉じたりしないことをおすすめします。
import { Sandbox, FilesystemEventType } from 'novita-sandbox/code-interpreter'
const sandbox = await Sandbox.create()
const dirname = '/tmp'
// Start watching directory for changes
const handle = await sandbox.files.watchDir(dirname, async (event) => {
console.log(`got event: ${event.type} - ${event.name}`)
if (event.type === FilesystemEventType.WRITE) {
console.log(`wrote to file ${event.name}`)
}
})
// Trigger file write event
await sandbox.files.write(`${dirname}/test-file`, 'test-file-content')
// Stop watching directory for changes
handle.stop()
await sandbox.kill()
from novita_sandbox.code_interpreter import Sandbox, FilesystemEventType
sandbox = Sandbox.create()
dirname = '/tmp'
# Watch directory for changes
handle = sandbox.files.watch_dir(dirname)
# Trigger file write event
sandbox.files.write(f"{dirname}/test-file", "test-file-content")
# Retrieve the latest new events since the last `get_new_events()` call
events = handle.get_new_events()
for event in events:
print(f"got event: {event.type} - {event.name}")
if event.type == FilesystemEventType.WRITE:
print(f"wrote to file {event.name}")
# Stop watching directory for changes
handle.stop()
sandbox.kill()
再帰的な監視
recursive パラメータを設定して、再帰的な監視を有効にできます。
新しいフォルダを短時間で連続して作成する場合(例: 深くネストされたフォルダのパス)、
CREATE 以外のイベントが発行されないことがあります。この動作を避けるには、必要なフォルダ構造を事前に作成してください。import { Sandbox, FilesystemEventType } from 'novita-sandbox/code-interpreter'
const sandbox = await Sandbox.create()
const dirname = '/tmp'
// Start watching directory for changes
const handle = await sandbox.files.watchDir(dirname, async (event) => {
console.log(`got event: ${event.type} - ${event.name}`)
if (event.type === FilesystemEventType.WRITE) {
console.log(`wrote to file ${event.name}`)
}
}, {
recursive: true
})
// Trigger file write event
await sandbox.files.write(`${dirname}/test-folder/test-file`, 'test-file-content')
// Stop watching directory for changes
handle.stop()
await sandbox.kill()
from novita_sandbox.code_interpreter import Sandbox, FilesystemEventType
sandbox = Sandbox.create()
dirname = '/tmp'
# Watch directory for changes
handle = sandbox.files.watch_dir(dirname, recursive=True)
# Trigger file write event
sandbox.files.write(f"{dirname}/test-folder/test-file", "test-file-content")
# Retrieve the latest new events since the last `get_new_events()` call
events = handle.get_new_events()
for event in events:
print(f"got event: {event.type} - {event.name}")
if event.type == FilesystemEventType.WRITE:
print(f"wrote to file {event.name}")
# Stop watching directory for changes
handle.stop()
sandbox.kill()