files.write() 方法向 Sandbox 上传数据。
在运行本文档中的示例代码之前,请确保已正确配置环境变量。详情请参阅”配置环境变量”。
上传单个文件
from ppio_sandbox import PPIO
ppio = PPIO()
sandbox = ppio.sandbox.create()
local_file_path = '../local-test-file'
with open(local_file_path, 'rb') as file:
file_path_in_sandbox = '/tmp/test-file'
result = sandbox.files.write(file_path_in_sandbox, file)
print(result)
sandbox.kill()
import fs from 'fs'
import { PPIO } from 'ppio-sandbox'
const ppio = new PPIO()
const sandbox = await ppio.sandbox.create()
// 从本地文件系统读取文件
const localFilePath = '../local-test-file'
const content = fs.readFileSync(localFilePath, 'utf8')
// 上传文件到 Sandbox
const filePathInSandbox = '/tmp/test-file'
const result = await sandbox.files.write(filePathInSandbox, content)
console.log(result)
await sandbox.kill()
使用预签名 URL 上传
预签名上传 URL 允许用户从不持有 PPIO SDK 凭证的环境(如 Web 浏览器)安全地上传文件。创建 Sandbox 时使用secure: true 选项,然后在你可信的后端使用 SDK 生成上传 URL。仅将其返回给你的应用授权的用户,并设置较短的过期时间。该 URL 在过期前相当于一个临时的 bearer 凭证。
from pathlib import Path
import requests
from ppio_sandbox import PPIO
ppio = PPIO()
sandbox = ppio.sandbox.create(secure=True)
local_file_path = Path('../local-test-file')
file_path_in_sandbox = '/tmp/test-file'
# 生成预签名上传 URL
signed_url = sandbox.upload_url(
path=file_path_in_sandbox,
use_signature_expiration=120, # 可选,单位为秒
)
# 模拟浏览器行为
with open(local_file_path, 'rb') as f:
response = requests.post(
signed_url,
files={'file': (local_file_path.name, f)},
timeout=60,
)
response.raise_for_status()
# 使用 SDK 回读验证
result = sandbox.files.read(file_path_in_sandbox)
print(result)
sandbox.kill()
import { PPIO } from 'ppio-sandbox'
const ppio = new PPIO()
const sandbox = await ppio.sandbox.create({ secure: true })
const filePathInSandbox = '/tmp/test-file'
// 生成预签名上传 URL
const publicUploadUrl = await sandbox.uploadUrl(filePathInSandbox, {
useSignatureExpiration: 120, // 可选,单位为秒
})
// 模拟浏览器行为
const form = new FormData()
form.append('file', new Blob(['file content']), 'test-file')
const uploadRes = await fetch(publicUploadUrl, {
method: 'POST',
body: form,
})
if (!uploadRes.ok) {
throw new Error(`Upload failed: ${uploadRes.status} ${await uploadRes.text()}`)
}
// 使用 SDK 回读验证
const result = await sandbox.files.read(filePathInSandbox)
console.log(result)
await sandbox.kill()
上传目录 / 多个文件
import os
from ppio_sandbox import PPIO
ppio = PPIO()
sandbox = ppio.sandbox.create()
def read_directory_files(directory_path):
files = []
for filename in os.listdir(directory_path):
file_path = os.path.join(directory_path, filename)
if os.path.isfile(file_path):
with open(file_path, 'rb') as file:
files.append({
'path': file_path,
'data': file.read(),
})
return files
local_directory_path = '../local-test-dir'
files = read_directory_files(local_directory_path)
print(files)
for file in files:
file['path'] = file['path'].replace('../local-test-dir', '/tmp')
result = sandbox.files.write_files(files)
print(result)
sandbox.kill()
import fs from 'fs'
import path from 'path'
import { PPIO } from 'ppio-sandbox'
const ppio = new PPIO()
const sandbox = await ppio.sandbox.create()
const readDirectoryFiles = (directoryPath) => {
const files = fs.readdirSync(directoryPath)
const filesArray = files
.filter((file) => {
const fullPath = path.join(directoryPath, file)
return fs.statSync(fullPath).isFile()
})
.map((file) => {
const filePath = path.join(directoryPath, file)
return {
path: filePath,
data: fs.readFileSync(filePath, 'utf8'),
}
})
return filesArray
}
const localDirectoryPath = '../local-test-dir'
const files = readDirectoryFiles(localDirectoryPath)
console.log(files)
files.forEach((file) => {
file.path = file.path.replace('../local-test-dir', '/tmp')
})
const result = await sandbox.files.write(files)
console.log(result)
await sandbox.kill()