Experimental snapshot sdk

examples/21_snapshot_and_spawn.py

@@ -0,0 +1,99 @@
+#!/usr/bin/env python3
+"""Create a sandbox, take a snapshot, and spawn a new sandbox from it"""
+
+import os
+import sys
+import random
+import string
+
+from koyeb import Sandbox
+from koyeb.sandbox import SnapshotType
+
+
+def main():
+    api_token = os.getenv("KOYEB_API_TOKEN")
+    if not api_token:
+        print("Error: KOYEB_API_TOKEN not set")
+        return 1
+
+    sbx = None
+    sbx2 = None
+    snapshot = None
+    suffix = "".join(random.choices(string.ascii_lowercase + string.digits, k=8))
+    try:
+        print("✓ Creating sandbox...")
+        sbx = Sandbox.create(
+            image="python:3.12",
+            name=f"snapshot-and-spawn-{suffix}",
+            wait_ready=True,
+            api_token=api_token,
+        )
+        print(f"  ✓ Sandbox created: {sbx.name}")
+
+        # Create some files to verify filesystem preservation
+        print("✓ Creating files...")
+        sbx.filesystem.mkdir("/workspace")
+        sbx.filesystem.write_file("/workspace/test_file.txt", "Hello from snapshot!")
+        sbx.filesystem.write_file("/workspace/requirements.txt", "requests\npytest\n")
+        print("  ✓ Files created")
+
+        # Install packages
+        print("✓ Installing packages...")
+        sbx.exec("pip3 install pytest requests")
+        print("  ✓ Packages installed")
+
+        # Take a snapshot
+        print("✓ Creating snapshot...")
+        snapshot = sbx.snapshot(
+            name=f"python-with-deps-{suffix}",
+            snapshot_type=SnapshotType.FILESYSTEM,
+        )
+        print(f"  ✓ Snapshot created: {snapshot.name}")
+
+        # Spawn a new sandbox from the snapshot with a different instance type
+        print("✓ Spawning sandbox from snapshot with different instance type...")
+        sbx2 = snapshot.spawn(
+            image="python:3.12",
+            name=f"test-runner-{suffix}",
+            instance_type="nano",
+            env={"SPAWNED_FROM_SNAPSHOT": "true"},
+            wait_ready=False,
+            api_token=api_token,
+        )
+        print(f"  ✓ Sandbox spawned: {sbx2.name}")
+
+        # Wait for the spawned sandbox to be ready
+        print("✓ Waiting for spawned sandbox to be ready...")
+        is_ready = sbx2.wait_ready(timeout=300)
+        print("  ✓ Sandbox is ready")
+
+        # Verify filesystem is preserved from snapshot
+        print("✓ Verifying filesystem is preserved from snapshot...")
+
+        # Check that files exist
+        result = sbx2.exec("cat /workspace/test_file.txt")
+        assert result.stdout.strip() == "Hello from snapshot!", f"File content mismatch: {result.stdout}"
+        print("  ✓ Custom file preserved")
+
+        # Check that packages are pre-installed
+        result = sbx2.exec("python3 -c \"import requests; print('OK')\"")
+        assert result.stdout.strip() == "OK", f"Package not found: {result.stdout}"
+        print("  ✓ Packages preserved")
+
+        # Verify env var is different from snapshot (new definition)
+        result = sbx2.exec("echo $SPAWNED_FROM_SNAPSHOT")
+        assert result.stdout.strip() == "true", f"Env var not set: {result.stdout}"
+        print("  ✓ New env vars applied (different from snapshot)")
+
+        return 0
+    finally:
+        if sbx2:
+            sbx2.delete()
+        if snapshot:
+            snapshot.delete()
+        if sbx:
+            sbx.delete()
+
+
+if __name__ == "__main__":
+    sys.exit(main())

examples/22_declarative_snapshot_builder.py

@@ -0,0 +1,92 @@
+#!/usr/bin/env python3
+"""Use the fluent builder API to create a reusable snapshot and spawn sandboxes from it"""
+
+import os
+import sys
+import random
+import string
+
+from koyeb import Sandbox
+
+
+def main():
+    api_token = os.getenv("KOYEB_API_TOKEN")
+    if not api_token:
+        print("Error: KOYEB_API_TOKEN not set")
+        return 1
+
+    snapshot = None
+    runner1 = None
+    runner2 = None
+    suffix = "".join(random.choices(string.ascii_lowercase + string.digits, k=8))
+    try:
+        # Build a reusable snapshot declaratively
+        print("✓ Building declarative snapshot...")
+        snapshot = (
+            Sandbox.template(
+                f"ci-environment-{suffix}",
+                image="python:3.12",
+                workdir="/workspace",
+                api_token=api_token,
+            )
+            .file("requirements.txt", "requests")
+            .run("pip install -r requirements.txt")
+            .build(snapshot_name=f"python-ci-env-{suffix}")
+        )
+
+        # Print operations performed during build
+        if snapshot.operations:
+            for op in snapshot.operations:
+                print(f"    - {op}")
+        print(f"  ✓ Snapshot built: {snapshot.name}")
+
+        # Spawn multiple sandboxes from the same snapshot in parallel
+        print("✓ Spawning sandboxes from snapshot...")
+        print("  - Spawning runner1...")
+        print("  - Spawning runner2...")
+        runner1 = snapshot.spawn(
+            image="python:3.12",
+            name=f"test-suite-1-{suffix}",
+            wait_ready=False,
+            api_token=api_token,
+        )
+        runner2 = snapshot.spawn(
+            image="python:3.12",
+            name=f"test-suite-2-{suffix}",
+            wait_ready=False,
+            api_token=api_token,
+        )
+        print(f"  ✓ Runner1 spawned: {runner1.name}")
+        print(f"  ✓ Runner2 spawned: {runner2.name}")
+
+        # Wait for both sandboxes to be ready
+        print("✓ Waiting for sandboxes to be ready...")
+        for runner in [runner1, runner2]:
+            runner.wait_ready(timeout=300)
+        print("  ✓ All sandboxes are ready")
+
+        # Verify the environment is pre-configured from snapshot filesystem
+        print("✓ Verifying runner1 has pre-installed packages from snapshot filesystem...")
+        result = runner1.exec("python3 -c \"import requests; print('All packages installed')\"")
+        print(f"  stdout: '{result.stdout}', stderr: '{result.stderr}', exit_code: {result.exit_code}")
+        assert "All packages installed" in result.stdout, f"Expected packages to be installed, got stdout: '{result.stdout}', stderr: '{result.stderr}'"
+        print(f"  ✓ Result: {result.stdout.strip()}")
+
+        print("✓ Verifying runner2 has pre-installed packages from snapshot filesystem...")
+        result = runner2.exec("python3 -c \"import requests; print('All packages installed')\"")
+        print(f"  stdout: '{result.stdout}', stderr: '{result.stderr}', exit_code: {result.exit_code}")
+        assert "All packages installed" in result.stdout, f"Expected packages to be installed, got stdout: '{result.stdout}', stderr: '{result.stderr}'"
+        print(f"  ✓ Result: {result.stdout.strip()}")
+
+        return 0
+    finally:
+        if runner2:
+            runner2.delete()
+        if runner1:
+            runner1.delete()
+        if snapshot:
+            snapshot.delete()
+
+
+if __name__ == "__main__":
+    sys.exit(main())

examples/23_full_snapshot_and_spawn.py

@@ -0,0 +1,105 @@
+#!/usr/bin/env python3
+"""Create a sandbox, take a full snapshot, and spawn a new sandbox from it
+
+This example demonstrates creating a full snapshot (including both filesystem
+and process state) as opposed to example 21 which uses filesystem-only snapshots.
+"""
+
+import os
+import sys
+import random
+import string
+
+from koyeb import Sandbox
+from koyeb.sandbox import SnapshotType
+from koyeb.sandbox.utils import SandboxTimeoutError
+
+
+def main():
+    api_token = os.getenv("KOYEB_API_TOKEN")
+    if not api_token:
+        print("Error: KOYEB_API_TOKEN not set")
+        return 1
+
+    sbx = None
+    sbx2 = None
+    snapshot = None
+    suffix = "".join(random.choices(string.ascii_lowercase + string.digits, k=8))
+    try:
+        print("✓ Creating sandbox...")
+        sbx = Sandbox.create(
+            image="python:3.12",
+            name=f"full-snapshot-and-spawn-{suffix}",
+            wait_ready=True,
+            api_token=api_token,
+        )
+        print(f"  ✓ Sandbox created: {sbx.name}")
+
+        # Create some files to verify filesystem preservation
+        print("✓ Creating files...")
+        sbx.filesystem.mkdir("/workspace")
+        sbx.filesystem.write_file("/workspace/requirements.txt", "requests\n")
+        print("  ✓ Files created")
+
+        # Install packages
+        print("✓ Installing packages...")
+        sbx.exec("pip3 install requests")
+        print("  ✓ Packages installed")
+
+        # Start daemon http server in the background on port 8000
+        print("✓ Running HTTP daemon server on port 8000...")
+        try:
+            sbx.exec("python3 -m http.server", timeout=2)
+        except SandboxTimeoutError:
+            pass
+        print("  ✓ Started HTTP daemon server on port 8000")
+
+        # Take a FULL snapshot (includes both filesystem and process state)
+        print("✓ Creating full snapshot...")
+        snapshot = sbx.snapshot(
+            name=f"python-full-snapshot-{suffix}",
+            snapshot_type=SnapshotType.FULL,
+        )
+        print(f"  ✓ Full snapshot created: {snapshot.name}")
+
+        # Spawn a new sandbox from the snapshot with a different instance type
+        print("✓ Spawning sandbox from full snapshot with different instance type...")
+        sbx2 = snapshot.spawn(
+            image="python:3.12",
+            name=f"test-runner-full-{suffix}",
+            instance_type="nano",
+            wait_ready=False,
+            api_token=api_token,
+        )
+        print(f"  ✓ Sandbox spawned: {sbx2.name}")
+
+        # Wait for the spawned sandbox to be ready
+        print("✓ Waiting for spawned sandbox to be ready...")
+        is_ready = sbx2.wait_ready(timeout=300)
+        print("  ✓ Sandbox is ready")
+
+        # Verify filesystem is preserved from snapshot
+        print("✓ Verifying full snapshot...")
+
+        # Check that packages are pre-installed
+        result = sbx2.exec("python3 -c \"import requests; print('OK')\"")
+        assert result.stdout.strip() == "OK", f"Package not found: {result.stdout}"
+        print("  ✓ Python package installed preserved")
+
+        # Verify processes are preserved
+        result = sbx2.exec("curl localhost:8000")
+        assert "Directory listing for /" in result.stdout.strip(), "HTTP server launched in original sandbox is not running anymore"
+        print("  ✓ Daemon HTTP server launched on initial sandbox is still running")
+
+        return 0
+    finally:
+        if sbx2:
+            sbx2.delete()
+        if snapshot:
+            snapshot.delete()
+        if sbx:
+            sbx.delete()
+
+
+if __name__ == "__main__":
+    sys.exit(main())