Migrating data between volumes

npm install e2b

pip install e2b

import { Sandbox, Volume } from 'e2b'

const [sourceName, destName] = process.argv.slice(2)
if (!sourceName || !destName) {
  throw new Error('Usage: copy-volume <source-volume> <dest-volume>')
}

async function findVolumeByName(name: string) {
  const volumes = await Volume.list()
  return volumes.find((v) => v.name === name) ?? null
}

const sourceInfo = await findVolumeByName(sourceName)
if (!sourceInfo) {
  throw new Error(`Source volume "${sourceName}" does not exist`)
}
const sourceVolume = await Volume.connect(sourceInfo.volumeId)

const destInfo = await findVolumeByName(destName)
const destVolume = destInfo
  ? await Volume.connect(destInfo.volumeId)
  : await Volume.create(destName)

function voltype(token: string): string {
  // JWT payloads are base64url and often omit padding, so normalize before decoding
  let segment = token.split('.')[1].replace(/-/g, '+').replace(/_/g, '/')
  segment += '='.repeat((4 - (segment.length % 4)) % 4)
  const payload = JSON.parse(atob(segment))
  return payload.voltype
}

console.log(`Source volume type:      ${voltype(sourceVolume.token)}`)
console.log(`Destination volume type: ${voltype(destVolume.token)}`)

const sandbox = await Sandbox.create({
  volumeMounts: {
    '/mnt/source': sourceVolume,
    '/mnt/dest': destVolume,
  },
})

try {
  const install = await sandbox.commands.run(
    'sudo apt-get update && sudo apt-get install -y rsync',
  )
  if (install.exitCode !== 0) {
    throw new Error(`rsync install failed: ${install.stderr}`)
  }

  const result = await sandbox.commands.run('rsync -a /mnt/source/ /mnt/dest/')
  if (result.exitCode !== 0) {
    throw new Error(`Copy failed: ${result.stderr}`)
  }

  const listing = await sandbox.commands.run('ls -la /mnt/dest')
  console.log(listing.stdout)
} finally {
  await sandbox.kill()
}

import base64
import json
import sys

from e2b import Sandbox, Volume

args = sys.argv[1:]
if len(args) != 2:
    raise SystemExit('Usage: copy_volume <source-volume> <dest-volume>')
source_name, dest_name = args


def find_volume_by_name(name):
    return next((v for v in Volume.list() if v.name == name), None)


def voltype(token):
    payload = token.split('.')[1]
    payload += '=' * (-len(payload) % 4)  # JWT segments omit base64 padding
    return json.loads(base64.urlsafe_b64decode(payload))['voltype']


source_info = find_volume_by_name(source_name)
if source_info is None:
    raise SystemExit(f'Source volume "{source_name}" does not exist')
source_volume = Volume.connect(source_info.volume_id)

dest_info = find_volume_by_name(dest_name)
dest_volume = (
    Volume.connect(dest_info.volume_id) if dest_info else Volume.create(dest_name)
)

print(f'Source volume type:      {voltype(source_volume.token)}')
print(f'Destination volume type: {voltype(dest_volume.token)}')

sandbox = Sandbox.create(
    volume_mounts={
        '/mnt/source': source_volume,
        '/mnt/dest': dest_volume,
    },
)

try:
    install = sandbox.commands.run(
        'sudo apt-get update && sudo apt-get install -y rsync',
    )
    if install.exit_code != 0:
        raise RuntimeError(f'rsync install failed: {install.stderr}')

    result = sandbox.commands.run('rsync -a /mnt/source/ /mnt/dest/')
    if result.exit_code != 0:
        raise RuntimeError(f'Copy failed: {result.stderr}')

    listing = sandbox.commands.run('ls -la /mnt/dest')
    print(listing.stdout)
finally:
    sandbox.kill()

export E2B_API_KEY=your-api-key
npx tsx copy-volume.mts prod-data prod-data-backup

export E2B_API_KEY=your-api-key
python copy_volume.py prod-data prod-data-backup