From f7150e1883670ca73bc23bebcd3155c584d178f8 Mon Sep 17 00:00:00 2001 From: nicoche <78445450+nicoche@users.noreply.github.com> Date: Mon, 20 Jul 2026 11:43:16 +0200 Subject: [PATCH 1/4] Wip backport --- examples/23_snapshot_and_spawn.py | 99 +++ examples/24_declarative_snapshot_builder.py | 92 +++ examples/25_full_snapshot_and_spawn.py | 105 ++++ examples/25_full_snapshot_and_spawn_async.py | 106 ++++ koyeb/sandbox/__init__.py | 9 + koyeb/sandbox/api.md | 103 ++++ koyeb/sandbox/sandbox.py | 606 ++++++++++++++++++- koyeb/sandbox/snapshot.py | 592 ++++++++++++++++++ koyeb/sandbox/utils.py | 12 + 9 files changed, 1709 insertions(+), 15 deletions(-) create mode 100644 examples/23_snapshot_and_spawn.py create mode 100644 examples/24_declarative_snapshot_builder.py create mode 100644 examples/25_full_snapshot_and_spawn.py create mode 100644 examples/25_full_snapshot_and_spawn_async.py create mode 100644 koyeb/sandbox/api.md create mode 100644 koyeb/sandbox/snapshot.py diff --git a/examples/23_snapshot_and_spawn.py b/examples/23_snapshot_and_spawn.py new file mode 100644 index 00000000..83c6e48a --- /dev/null +++ b/examples/23_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()) diff --git a/examples/24_declarative_snapshot_builder.py b/examples/24_declarative_snapshot_builder.py new file mode 100644 index 00000000..2f8a6c8d --- /dev/null +++ b/examples/24_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()) diff --git a/examples/25_full_snapshot_and_spawn.py b/examples/25_full_snapshot_and_spawn.py new file mode 100644 index 00000000..7e1dd6ab --- /dev/null +++ b/examples/25_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()) diff --git a/examples/25_full_snapshot_and_spawn_async.py b/examples/25_full_snapshot_and_spawn_async.py new file mode 100644 index 00000000..3e78fe78 --- /dev/null +++ b/examples/25_full_snapshot_and_spawn_async.py @@ -0,0 +1,106 @@ +#!/usr/bin/env python3 +"""Create a sandbox, take a full snapshot, and spawn a new sandbox from it (async version) + +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 asyncio +import random +import string + +from koyeb import AsyncSandbox +from koyeb.sandbox import SnapshotType +from koyeb.sandbox.utils import SandboxTimeoutError + + +async 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 = await AsyncSandbox.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...") + await sbx.filesystem.mkdir("/workspace") + await sbx.filesystem.write_file("/workspace/requirements.txt", "requests\n") + print(" ✓ Files created") + + # Install packages + print("✓ Installing packages...") + await 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: + await 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 = await 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 filesystem is preserved from 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(" ✓ Packages 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: + await sbx.delete() + + +if __name__ == "__main__": + sys.exit(asyncio.run(main())) \ No newline at end of file diff --git a/koyeb/sandbox/__init__.py b/koyeb/sandbox/__init__.py index 7fb74fd6..c1dac153 100644 --- a/koyeb/sandbox/__init__.py +++ b/koyeb/sandbox/__init__.py @@ -19,12 +19,17 @@ ) from .filesystem import FileInfo, SandboxFilesystem from .sandbox import AsyncSandbox, ExposedPort, ProcessInfo, Sandbox +from .snapshot import DeclarativeSnapshot, Snapshot, SnapshotStatus, SnapshotType from .utils import ( EgressPolicyError, SandboxDeploymentError, SandboxError, SandboxServiceError, SandboxTimeoutError, + SandboxDeploymentError, + SandboxError, + SandboxServiceError, + SandboxTimeoutError ) __all__ = [ @@ -47,4 +52,8 @@ "SandboxCommandError", "ExposedPort", "ProcessInfo", + "Snapshot", + "SnapshotType", + "SnapshotStatus", + "DeclarativeSnapshot", ] diff --git a/koyeb/sandbox/api.md b/koyeb/sandbox/api.md new file mode 100644 index 00000000..7cbe8081 --- /dev/null +++ b/koyeb/sandbox/api.md @@ -0,0 +1,103 @@ +# Sandbox Snapshot API + +## API Surface + +```python +# Snapshot +class Snapshot: + @classmethod + def get(snapshot_id, api_token=None, host=None) -> Snapshot + @classmethod + def list(service_id=None, snapshot_type=None, status=None, limit=50, offset=0, api_token=None, host=None) -> List[Snapshot] + def refresh() -> None + def wait_available(timeout=600, poll_interval=5.0) -> bool + def delete() -> bool + def spawn(name=None, wait_ready=True, timeout=300, **kwargs) -> Sandbox + +# DeclarativeSnapshot +class DeclarativeSnapshot: + def __init__(name, image, workdir=None, api_token=None, host=None, delete_builder=True) + # Declare build steps. + def file(path, content) -> DeclarativeSnapshot + def copy(src, dst) -> DeclarativeSnapshot + def run(command, cwd=None) -> DeclarativeSnapshot + # Run the build steps and generate a snapshot. + def build(snapshot_name=None) -> Snapshot + +# Sandbox extensions +class Sandbox: + def snapshot(name, snapshot_type=SnapshotType.FILESYSTEM, wait_available=True, timeout=600) -> Snapshot + @staticmethod + def template(name, image, workdir=None, api_token=None, host=None, delete_builder=True) -> DeclarativeSnapshot +``` + +## Enums + +**SnapshotType**: `FILESYSTEM` | `FULL` +**SnapshotStatus**: `INVALID` | `CREATING` | `UPLOADING` | `AVAILABLE` | `DELETING` | `DELETED` | `FAILED` + +## Snapshot + +### Fields +`id`, `name`, `service_id`, `snapshot_type`, `status`, `created_at`, `deployment_id`, `available_at`, `size`, `region`, `organization_id`, `project_id`, `messages`, `operations`, `api_token`, `host`, `sandbox_secret` + +### Methods +`get(snapshot_id, api_token, host) -> Snapshot` +`list(service_id, snapshot_type, status, limit, offset, api_token, host) -> List[Snapshot]` +`refresh()` +`wait_available(timeout=600, poll_interval=5.0) -> bool` +`delete() -> bool` +`spawn(name, wait_ready=True, timeout=300, **kwargs) -> Sandbox` + +## Sandbox + +`snapshot(name, snapshot_type=FILESYSTEM, wait_available=True, timeout=600) -> Snapshot` +`template(name, image, workdir=None, api_token=None, host=None, delete_builder=True) -> DeclarativeSnapshot` + +## DeclarativeSnapshot + +Builder for creating snapshots declaratively. + +`file(path, content) -> DeclarativeSnapshot` +`copy(src, dst) -> DeclarativeSnapshot` +`run(command, cwd=None) -> DeclarativeSnapshot` +`build(snapshot_name=None) -> Snapshot` - Returns snapshot with operations list +`get_operations() -> List[str]` + +## Examples + +### 1. Snapshot & Spawn +```python +from koyeb import Sandbox +sbx = Sandbox.create(image="python:3.12", wait_ready=True) +sbx.exec("pip3 install requests") +snapshot = sbx.snapshot(name="python-with-deps") +sbx2 = snapshot.spawn(name="test-runner", wait_ready=True) +``` + +### 2. Declarative Builder +```python +from koyeb import Sandbox +snapshot = ( + Sandbox.template("ci-env", image="python:3.12", workdir="/workspace") + .file("requirements.txt", "pytest\nrequests") + .run("pip install -r requirements.txt") + .build(snapshot_name="ci-snapshot") +) +for op in snapshot.operations: + print(f" - {op}") +``` + +### 3. Full Snapshot +```python +from koyeb.sandbox import SnapshotType +snapshot = sbx.snapshot(name="full", snapshot_type=SnapshotType.FULL) +``` + +### 4. Parallel Spawn +```python +runner1 = snapshot.spawn(name="runner1", wait_ready=False) +runner2 = snapshot.spawn(name="runner2", wait_ready=False) +for runner in [runner1, runner2]: + runner.wait_ready(timeout=300) +``` diff --git a/koyeb/sandbox/sandbox.py b/koyeb/sandbox/sandbox.py index d3b9c6f9..d325281c 100644 --- a/koyeb/sandbox/sandbox.py +++ b/koyeb/sandbox/sandbox.py @@ -11,7 +11,8 @@ import secrets import time from dataclasses import dataclass -from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple +from datetime import datetime +from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Union from koyeb.api.api.deployments_api import DeploymentsApi from koyeb.api.exceptions import ApiException, NotFoundException @@ -46,6 +47,7 @@ from .exec import AsyncSandboxExecutor, SandboxExecutor from .executor_client import AsyncSandboxClient, SandboxClient from .filesystem import AsyncSandboxFilesystem, SandboxFilesystem + from .snapshot import SnapshotType @dataclass @@ -90,6 +92,7 @@ def __init__( sandbox_secret: Optional[str] = None, poll_interval: float = DEFAULT_POLL_INTERVAL, host: Optional[str] = None, + snapshot_id: Optional[str] = None, ): self.sandbox_id = sandbox_id self.app_id = app_id @@ -99,6 +102,7 @@ def __init__( self.sandbox_secret = sandbox_secret self.poll_interval = poll_interval self.host = host + self.snapshot_id = snapshot_id self._created_at = time.time() self._sandbox_url: Optional[Tuple[str, Optional[str]]] = None self._domain: Optional[str] = None @@ -143,6 +147,8 @@ def create( host: Optional[str] = None, block_network: bool = False, outbound_allowlist: Optional[List[str]] = None, + snapshot: Optional[Union[str, "Snapshot"]] = None, + sandbox_secret: Optional[str] = None, ) -> Sandbox: """ Create a new sandbox instance. @@ -188,6 +194,10 @@ def create( outbound_allowlist: List of IPs/CIDRs allowed as outbound destinations; all other outbound traffic is blocked. Bare IPs are normalized to /32 (IPv4) or /128 (IPv6). Mutually exclusive with block_network. + snapshot: Optional. A Snapshot object or snapshot name/ID string to create the sandbox from. + If provided, the sandbox will be initialized from this snapshot. + Can be either a Snapshot object (e.g., snapshot=my_snapshot) or a snapshot name/ID string (e.g., snapshot="my snapshot"). + sandbox_secret: Optional sandbox secret to use for executor authentication. If not provided, a new one will be generated. Returns: Sandbox: A new Sandbox instance @@ -207,6 +217,22 @@ def create( ... image="ghcr.io/myorg/myimage:latest", ... registry_secret="my-ghcr-secret" ... ) + + >>> # Create from a Snapshot object + >>> from koyeb.sandbox import Snapshot + >>> snapshot = Snapshot.get("my-snapshot-id") + >>> sandbox = Sandbox.create(snapshot=snapshot) + + >>> # Create from a snapshot ID string + >>> sandbox = Sandbox.create(snapshot="my-snapshot-id") + + >>> # Create from a snapshot with custom parameters + >>> sandbox = Sandbox.create( + ... snapshot="my-snapshot-id", + ... image="python:3.12", + ... instance_type="nano", + ... env={"MY_VAR": "value"} + ... ) """ if api_token is None: api_token = os.getenv("KOYEB_API_TOKEN") @@ -215,6 +241,43 @@ def create( "API token is required. Set KOYEB_API_TOKEN environment variable or pass api_token parameter" ) + # Handle snapshot parameter (can be Snapshot object or snapshot name/ID string) + actual_snapshot_id = None + actual_snapshot_type = None + + if snapshot is not None: + if isinstance(snapshot, str): + # snapshot is a snapshot ID or name string + from .snapshot import Snapshot + + # Try to get snapshot by ID first (fast path for UUIDs) + try: + snapshot_obj = Snapshot.get(snapshot, api_token=api_token, host=host) + actual_snapshot_id = snapshot_obj.id + actual_snapshot_type = snapshot_obj.snapshot_type + except Exception: + # If that fails, try to find by name + try: + snapshots = Snapshot.list(api_token=api_token, host=host, limit=100) + for s in snapshots: + if s.name == snapshot: + actual_snapshot_id = s.id + actual_snapshot_type = s.snapshot_type + break + except Exception: + pass + + # If we couldn't resolve it, use the string as-is + # Default to FILESYSTEM type if we couldn't determine it + if actual_snapshot_id is None: + actual_snapshot_id = snapshot + from .snapshot import SnapshotType + actual_snapshot_type = SnapshotType.FILESYSTEM + else: + # snapshot is a Snapshot object + actual_snapshot_id = snapshot.id + actual_snapshot_type = snapshot.snapshot_type + sandbox = cls._create_sync( name=name, image=image, @@ -242,6 +305,9 @@ def create( host=host, block_network=block_network, outbound_allowlist=outbound_allowlist, + snapshot_id=actual_snapshot_id, + snapshot_type=actual_snapshot_type, + sandbox_secret=sandbox_secret, ) if wait_ready: @@ -284,6 +350,9 @@ def _create_sync( host: Optional[str] = None, block_network: bool = False, outbound_allowlist: Optional[List[str]] = None, + snapshot_id: Optional[str] = None, + snapshot_type: Optional["SnapshotType"] = None, + sandbox_secret: Optional[str] = None, ) -> Sandbox: """ Synchronous creation method that returns creation parameters. @@ -298,8 +367,9 @@ def _create_sync( # Always create routes (ports are always exposed, default to "http") routes = create_koyeb_sandbox_routes() - # Generate secure sandbox secret - sandbox_secret = secrets.token_urlsafe(32) + # Generate secure sandbox secret if not provided + if sandbox_secret is None: + sandbox_secret = secrets.token_urlsafe(32) # Add SANDBOX_SECRET to environment variables if env is None: @@ -348,11 +418,83 @@ def _create_sync( delete_after_create=delete_after_delay, delete_after_sleep=delete_after_inactivity_delay, ) - create_service = CreateService( - app_id=app_id, - definition=deployment_definition, - life_cycle=service_life_cycle, - ) + + # Handle snapshot creation based on snapshot type + # For FULL snapshots, don't provide definition (API will infer it) + # For FILESYSTEM snapshots, always provide definition with snapshot_id + if snapshot_id: + # Import here to avoid circular import + from .snapshot import SnapshotType as ST + + # For FULL snapshots, create service without definition + if snapshot_type == ST.FULL and snapshot_type is not None: + create_service = CreateService( + app_id=app_id, + life_cycle=service_life_cycle, + instance_snapshot_id=snapshot_id, + name=name, + ) + else: + # For FILESYSTEM snapshots (or unknown), provide definition + env_vars = build_env_vars(env) + config_file_objects = build_config_files(config_files) + docker_source = create_docker_source( + image, privileged=privileged, image_registry_secret=registry_secret, + entrypoint=entrypoint, command=command, args=args, + ) + deployment_definition = create_deployment_definition( + name=name, + docker_source=docker_source, + env_vars=env_vars, + instance_type=instance_type, + exposed_port_protocol=exposed_port_protocol, + region=region, + routes=routes, + idle_timeout=idle_timeout, + enable_tcp_proxy=enable_tcp_proxy, + _experimental_enable_light_sleep=_experimental_enable_light_sleep, + _experimental_deep_sleep_value=_experimental_deep_sleep_value, + enable_mesh=enable_mesh, + config_files=config_file_objects if config_file_objects else None, + network_policy=network_policy, + ) + create_service = CreateService( + app_id=app_id, + definition=deployment_definition, + life_cycle=service_life_cycle, + instance_snapshot_id=snapshot_id, + name=name, + ) + else: + # No snapshot, create normally with definition + env_vars = build_env_vars(env) + config_file_objects = build_config_files(config_files) + docker_source = create_docker_source( + image, privileged=privileged, image_registry_secret=registry_secret, + entrypoint=entrypoint, command=command, args=args, + ) + deployment_definition = create_deployment_definition( + name=name, + docker_source=docker_source, + env_vars=env_vars, + instance_type=instance_type, + exposed_port_protocol=exposed_port_protocol, + region=region, + routes=routes, + idle_timeout=idle_timeout, + enable_tcp_proxy=enable_tcp_proxy, + _experimental_enable_light_sleep=_experimental_enable_light_sleep, + _experimental_deep_sleep_value=_experimental_deep_sleep_value, + enable_mesh=enable_mesh, + config_files=config_file_objects if config_file_objects else None, + network_policy=network_policy, + ) + create_service = CreateService( + app_id=app_id, + definition=deployment_definition, + life_cycle=service_life_cycle, + name=name, + ) service_response = services_api.create_service(service=create_service) service_id = service_response.service.id @@ -365,6 +507,7 @@ def _create_sync( sandbox_secret=sandbox_secret, poll_interval=poll_interval, host=host, + snapshot_id=snapshot_id, ) @classmethod @@ -470,6 +613,188 @@ def get_from_id( return sandbox + def snapshot( + self, + name: str, + snapshot_type: "SnapshotType" = None, + wait_available: bool = True, + timeout: int = 600, + ) -> "Snapshot": + """ + Create a snapshot of this sandbox. + + Captures the current state of the sandbox's filesystem (and optionally + running processes for FULL type) so it can be restored later. + + Args: + name: Name for the snapshot + snapshot_type: Type of snapshot to create (FILESYSTEM or FULL). + Defaults to FILESYSTEM. + wait_available: Whether to wait for snapshot to become available + timeout: Timeout in seconds for waiting + + Returns: + Snapshot: The created snapshot object + + Raises: + SandboxError: If snapshot creation fails + """ + from .snapshot import Snapshot, SnapshotType + + if snapshot_type is None: + snapshot_type = SnapshotType.FILESYSTEM + + try: + from koyeb.api.models.instance_snapshot_type import InstanceSnapshotType + from koyeb.api.models.create_instance_snapshot_request import CreateInstanceSnapshotRequest + from .snapshot import SnapshotStatus + + # Get API clients + clients = get_api_clients(self.api_token, self.host) + + from koyeb.api.models.instance_status import InstanceStatus + + # Get the first running instance for this service + instances_reply = clients.instances.list_instances( + service_id=self.service_id, + statuses=[InstanceStatus.HEALTHY, InstanceStatus.STARTING, InstanceStatus.ALLOCATING], + limit="1", + ) + + if not instances_reply.instances or len(instances_reply.instances) == 0: + raise SandboxError(f"No running instances found for service {self.service_id}") + + instance = instances_reply.instances[0] + instance_id = instance.id + + # Map SnapshotType to InstanceSnapshotType + if snapshot_type == SnapshotType.FILESYSTEM: + instance_snapshot_type = InstanceSnapshotType.INSTANCE_SNAPSHOT_TYPE_FILESYSTEM + else: + instance_snapshot_type = InstanceSnapshotType.INSTANCE_SNAPSHOT_TYPE_FULL + + # Create the snapshot via API + create_request = CreateInstanceSnapshotRequest( + instance_id=instance_id, + name=name, + type=instance_snapshot_type, + ) + + reply = clients.instance_snapshots.create_instance_snapshot(create_request) + + if not reply.instance_snapshot: + raise SandboxError("Failed to create snapshot: no snapshot returned from API") + + api_snapshot = reply.instance_snapshot + + # Convert API snapshot to our Snapshot object + # Include the sandbox_secret so spawned sandboxes can use the same executor + snapshot = Snapshot._from_instance_api_snapshot( + api_snapshot, self.api_token, self.host, sandbox_secret=self.sandbox_secret + ) + + # Wait for snapshot to become available if requested + if wait_available: + start_time = time.time() + while time.time() - start_time < timeout: + snapshot.refresh() + if snapshot.status == SnapshotStatus.AVAILABLE: + break + if snapshot.status == SnapshotStatus.FAILED: + raise SandboxError(f"Snapshot creation failed: {snapshot.messages}") + time.sleep(self.poll_interval) + else: + raise SandboxTimeoutError( + f"Snapshot did not become available within {timeout} seconds" + ) + + return snapshot + + except Exception as e: + raise SandboxError(f"Failed to create snapshot: {e}") from e + + @classmethod + def create_from_snapshot( + cls, + snapshot: Union["Snapshot", str], + name: Optional[str] = None, + wait_ready: bool = True, + timeout: int = 300, + **create_kwargs, + ) -> "Sandbox": + """ + Create a new sandbox from a snapshot. + + Args: + snapshot: Snapshot object or snapshot ID string + name: Name for the new sandbox + wait_ready: Whether to wait for sandbox to be ready + timeout: Timeout in seconds + **create_kwargs: Additional arguments to pass to create() + + Returns: + Sandbox: A new sandbox instance + """ + create_params = { + "snapshot": snapshot, + "wait_ready": wait_ready, + "timeout": timeout, + } + if name: + create_params["name"] = name + + create_params.update(create_kwargs) + return cls.create(**create_params) + + @classmethod + def template( + cls, + name: str, + image: str, + workdir: Optional[str] = None, + api_token: Optional[str] = None, + host: Optional[str] = None, + delete_builder: bool = True, + ) -> "DeclarativeSnapshot": + """ + Create a declarative snapshot builder. + + Use this to build a reusable snapshot by declaratively defining + the sandbox environment (files, packages, etc.) and then building + a snapshot that can be used to spawn pre-configured sandboxes. + + Args: + name: Name for the template + image: Docker image to use + workdir: Working directory in the sandbox + api_token: Koyeb API token + host: Koyeb API host + delete_builder: Whether to delete the builder sandbox after creating the snapshot (default: True) + + Returns: + DeclarativeSnapshot: Fluent builder for creating snapshots + + Example: + snapshot = ( + Sandbox.template("python-ci", image="python:3.12", workdir="/workspace") + .file("requirements.txt", "pytest\\nrequests") + .run("pip install -r requirements.txt") + .build(snapshot_name="python-ci-env") + ) + + sbx = snapshot.spawn(name="test-runner") + """ + from .snapshot import DeclarativeSnapshot + + return DeclarativeSnapshot( + name=name, + image=image, + workdir=workdir, + api_token=api_token, + host=host, + delete_builder=delete_builder, + ) + _DEPLOYMENT_ERROR_STATUSES = { DeploymentStatus.ERROR, DeploymentStatus.ERRORING, @@ -1351,6 +1676,8 @@ async def create( host: Optional[str] = None, block_network: bool = False, outbound_allowlist: Optional[List[str]] = None, + snapshot: Optional[Union[str, "Snapshot"]] = None, + sandbox_secret: Optional[str] = None, ) -> AsyncSandbox: """ Create a new sandbox instance with async support. @@ -1398,6 +1725,10 @@ async def create( outbound_allowlist: List of IPs/CIDRs allowed as outbound destinations; all other outbound traffic is blocked. Bare IPs are normalized to /32 (IPv4) or /128 (IPv6). Mutually exclusive with block_network. + snapshot: Optional. A Snapshot object or snapshot name/ID string to create the sandbox from. + If provided, the sandbox will be initialized from this snapshot. + Can be either a Snapshot object (e.g., snapshot=my_snapshot) or a snapshot name/ID string (e.g., snapshot="my snapshot"). + sandbox_secret: Optional sandbox secret to use for executor authentication. If not provided, a new one will be generated. Returns: AsyncSandbox: A new AsyncSandbox instance @@ -1415,6 +1746,43 @@ async def create( "API token is required. Set KOYEB_API_TOKEN environment variable or pass api_token parameter" ) + # Handle snapshot parameter (can be Snapshot object or snapshot name/ID string) + actual_snapshot_id = None + actual_snapshot_type = None + + if snapshot is not None: + if isinstance(snapshot, str): + # snapshot is a snapshot ID or name string + from .snapshot import Snapshot + + # Try to get snapshot by ID first (fast path for UUIDs) + try: + snapshot_obj = Snapshot.get(snapshot, api_token=api_token, host=host) + actual_snapshot_id = snapshot_obj.id + actual_snapshot_type = snapshot_obj.snapshot_type + except Exception: + # If that fails, try to find by name + try: + snapshots = Snapshot.list(api_token=api_token, host=host, limit=100) + for s in snapshots: + if s.name == snapshot: + actual_snapshot_id = s.id + actual_snapshot_type = s.snapshot_type + break + except Exception: + pass + + # If we couldn't resolve it, use the string as-is + # Default to FILESYSTEM type if we couldn't determine it + if actual_snapshot_id is None: + actual_snapshot_id = snapshot + from .snapshot import SnapshotType + actual_snapshot_type = SnapshotType.FILESYSTEM + else: + # snapshot is a Snapshot object + actual_snapshot_id = snapshot.id + actual_snapshot_type = snapshot.snapshot_type + from .utils import get_async_api_clients, build_network_policy from koyeb.api_async.models.create_app import CreateApp as AsyncCreateApp from koyeb.api_async.models.create_app import AppLifeCycle as AsyncAppLifeCycle @@ -1430,8 +1798,9 @@ async def create( # Always create routes routes = create_koyeb_sandbox_routes() - # Generate secure sandbox secret - sandbox_secret = secrets.token_urlsafe(32) + # Generate secure sandbox secret if not provided + if sandbox_secret is None: + sandbox_secret = secrets.token_urlsafe(32) # Add SANDBOX_SECRET to environment variables if env is None: @@ -1478,11 +1847,83 @@ async def create( ) # Convert sync DeploymentDefinition to dict so the async Pydantic model # (which expects koyeb.api_async.models.DeploymentDefinition) can coerce it. - create_service = AsyncCreateService( - app_id=app_id, - definition=deployment_definition.to_dict(), - life_cycle=service_life_cycle, - ) + + # Handle snapshot creation based on snapshot type + # For FULL snapshots, don't provide definition (API will infer it) + # For FILESYSTEM snapshots, always provide definition with snapshot_id + if actual_snapshot_id: + # Import here to avoid circular import + from .snapshot import SnapshotType as ST + + # For FULL snapshots, create service without definition + if actual_snapshot_type == ST.FULL: + create_service = AsyncCreateService( + app_id=app_id, + life_cycle=service_life_cycle, + instance_snapshot_id=actual_snapshot_id, + name=name, + ) + else: + # For FILESYSTEM snapshots (or unknown), provide definition + env_vars = build_env_vars(env) + config_file_objects = build_config_files(config_files) + docker_source = create_docker_source( + image, privileged=privileged, image_registry_secret=registry_secret, + entrypoint=entrypoint, command=command, args=args, + ) + deployment_definition = create_deployment_definition( + name=name, + docker_source=docker_source, + env_vars=env_vars, + instance_type=instance_type, + exposed_port_protocol=exposed_port_protocol, + region=region, + routes=routes, + idle_timeout=idle_timeout, + enable_tcp_proxy=enable_tcp_proxy, + _experimental_enable_light_sleep=_experimental_enable_light_sleep, + _experimental_deep_sleep_value=_experimental_deep_sleep_value, + enable_mesh=enable_mesh, + config_files=config_file_objects if config_file_objects else None, + network_policy=network_policy, + ) + create_service = AsyncCreateService( + app_id=app_id, + definition=deployment_definition.to_dict(), + life_cycle=service_life_cycle, + instance_snapshot_id=actual_snapshot_id, + name=name, + ) + else: + # No snapshot, create normally with definition + env_vars = build_env_vars(env) + config_file_objects = build_config_files(config_files) + docker_source = create_docker_source( + image, privileged=privileged, image_registry_secret=registry_secret, + entrypoint=entrypoint, command=command, args=args, + ) + deployment_definition = create_deployment_definition( + name=name, + docker_source=docker_source, + env_vars=env_vars, + instance_type=instance_type, + exposed_port_protocol=exposed_port_protocol, + region=region, + routes=routes, + idle_timeout=idle_timeout, + enable_tcp_proxy=enable_tcp_proxy, + _experimental_enable_light_sleep=_experimental_enable_light_sleep, + _experimental_deep_sleep_value=_experimental_deep_sleep_value, + enable_mesh=enable_mesh, + config_files=config_file_objects if config_file_objects else None, + network_policy=network_policy, + ) + create_service = AsyncCreateService( + app_id=app_id, + definition=deployment_definition.to_dict(), + life_cycle=service_life_cycle, + name=name, + ) service_response = await clients.services.create_service(service=create_service) service_id = service_response.service.id @@ -1495,6 +1936,7 @@ async def create( sandbox_secret=sandbox_secret, poll_interval=poll_interval, host=host, + snapshot_id=actual_snapshot_id, ) if wait_ready: @@ -1664,6 +2106,140 @@ async def delete(self) -> None: clients = get_async_api_clients(self.api_token, self.host) await clients.apps.delete_app(self.app_id) + async def snapshot( + self, + name: str, + snapshot_type: "SnapshotType" = None, + wait_available: bool = True, + timeout: int = 600, + ) -> "Snapshot": + """ + Create a snapshot of this sandbox asynchronously. + + Captures the current state of the sandbox's filesystem (and optionally + running processes for FULL type) so it can be restored later. + + Args: + name: Name for the snapshot + snapshot_type: Type of snapshot to create (FILESYSTEM or FULL). + Defaults to FILESYSTEM. + wait_available: Whether to wait for snapshot to become available + timeout: Timeout in seconds for waiting + + Returns: + Snapshot: The created snapshot object + + Raises: + SandboxError: If snapshot creation fails + """ + from .snapshot import Snapshot, SnapshotType + + if snapshot_type is None: + snapshot_type = SnapshotType.FILESYSTEM + + try: + from koyeb.api_async.models.instance_snapshot_type import InstanceSnapshotType + from koyeb.api_async.models.create_instance_snapshot_request import CreateInstanceSnapshotRequest + from .utils import get_async_api_clients + from .snapshot import SnapshotStatus + + # Get async API clients + clients = get_async_api_clients(self.api_token, self.host) + + from koyeb.api.models.instance_status import InstanceStatus + + # Get the first running instance for this service + instances_reply = await clients.instances.list_instances( + service_id=self.service_id, + statuses=[InstanceStatus.HEALTHY, InstanceStatus.STARTING, InstanceStatus.ALLOCATING], + limit="1", + ) + + if not instances_reply.instances or len(instances_reply.instances) == 0: + raise SandboxError(f"No running instances found for service {self.service_id}") + + instance = instances_reply.instances[0] + instance_id = instance.id + + # Map SnapshotType to InstanceSnapshotType + if snapshot_type == SnapshotType.FILESYSTEM: + instance_snapshot_type = InstanceSnapshotType.INSTANCE_SNAPSHOT_TYPE_FILESYSTEM + else: + instance_snapshot_type = InstanceSnapshotType.INSTANCE_SNAPSHOT_TYPE_FULL + + # Create the snapshot via API + create_request = CreateInstanceSnapshotRequest( + instance_id=instance_id, + name=name, + type=instance_snapshot_type, + ) + + reply = await clients.instance_snapshots.create_instance_snapshot(create_request) + + if not reply.instance_snapshot: + raise SandboxError("Failed to create snapshot: no snapshot returned from API") + + api_snapshot = reply.instance_snapshot + + # Convert API snapshot to our Snapshot object + # Include the sandbox_secret so spawned sandboxes can use the same executor + snapshot = Snapshot._from_instance_api_snapshot( + api_snapshot, self.api_token, self.host, sandbox_secret=self.sandbox_secret + ) + + # Wait for snapshot to become available if requested + if wait_available: + start_time = time.time() + while time.time() - start_time < timeout: + snapshot.refresh() + if snapshot.status == SnapshotStatus.AVAILABLE: + break + if snapshot.status == SnapshotStatus.FAILED: + raise SandboxError(f"Snapshot creation failed: {snapshot.messages}") + await asyncio.sleep(self.poll_interval) + else: + raise SandboxTimeoutError( + f"Snapshot did not become available within {timeout} seconds" + ) + + return snapshot + + except Exception as e: + raise SandboxError(f"Failed to create snapshot: {e}") from e + + @classmethod + async def create_from_snapshot( + cls, + snapshot: Union["Snapshot", str], + name: Optional[str] = None, + wait_ready: bool = True, + timeout: int = 300, + **create_kwargs, + ) -> "AsyncSandbox": + """ + Create a new async sandbox from a snapshot. + + Args: + snapshot: Snapshot object or snapshot ID string + name: Name for the new sandbox + wait_ready: Whether to wait for sandbox to be ready + timeout: Timeout in seconds + **create_kwargs: Additional arguments to pass to create() + + Returns: + AsyncSandbox: A new async sandbox instance + """ + create_params = { + "snapshot": snapshot, + "wait_ready": wait_ready, + "timeout": timeout, + } + if name: + create_params["name"] = name + + create_params.update(create_kwargs) + return await cls.create(**create_params) + async def is_healthy(self) -> bool: """Check if sandbox is healthy and ready for operations asynchronously.""" if not await self._async_is_deployment_healthy(): diff --git a/koyeb/sandbox/snapshot.py b/koyeb/sandbox/snapshot.py new file mode 100644 index 00000000..00160c5c --- /dev/null +++ b/koyeb/sandbox/snapshot.py @@ -0,0 +1,592 @@ +# coding: utf-8 + +""" +Koyeb Sandbox Snapshot - Snapshot functionality for Koyeb sandboxes +""" + +from __future__ import annotations + +import time +from dataclasses import dataclass, field +from datetime import datetime +from enum import Enum +from typing import Any, ClassVar, Dict, List, Optional, Union + +from .sandbox import Sandbox +from .utils import SandboxError, get_api_clients + + +class SnapshotType(Enum): + """Types of sandbox snapshots.""" + + FILESYSTEM = "filesystem" + FULL = "full" + + +class SnapshotStatus(Enum): + """Status of a sandbox snapshot.""" + + INVALID = "invalid" + CREATING = "creating" + UPLOADING = "uploading" + AVAILABLE = "available" + DELETING = "deleting" + DELETED = "deleted" + FAILED = "failed" + + +@dataclass +class Snapshot: + """ + Represents a sandbox snapshot resource. + + A snapshot captures the state of a sandbox at a specific point in time, + including its filesystem and optionally running processes. Sandboxes can + be spawned from snapshots to create pre-configured environments. + """ + + id: str + name: str + service_id: str + snapshot_type: SnapshotType + status: SnapshotStatus + created_at: datetime + deployment_id: Optional[str] = None + available_at: Optional[datetime] = None + size: Optional[int] = None + region: str = "" + organization_id: str = "" + project_id: Optional[str] = None + messages: List[str] = field(default_factory=list) + api_token: Optional[str] = None + host: Optional[str] = None + sandbox_secret: Optional[str] = None + operations: List[str] = field(default_factory=list) + + @classmethod + def get( + cls, + snapshot_id: str, + api_token: Optional[str] = None, + host: Optional[str] = None, + ) -> Snapshot: + """ + Get a snapshot by ID. + + Uses the InstanceSnapshots API which is for sandbox/service instance snapshots. + + Args: + snapshot_id: The ID of the snapshot to retrieve + api_token: Koyeb API token (falls back to KOYEB_API_TOKEN env var) + host: Koyeb API host + + Returns: + Snapshot: The snapshot object + + Raises: + SandboxError: If snapshot cannot be retrieved + """ + import os + + if not api_token: + api_token = os.getenv("KOYEB_API_TOKEN") + if not api_token: + raise SandboxError("API token is required. Set KOYEB_API_TOKEN environment variable.") + + try: + clients = get_api_clients(api_token, host) + reply = clients.instance_snapshots.get_instance_snapshot(snapshot_id) + + if not reply.instance_snapshot: + raise SandboxError(f"Snapshot {snapshot_id} not found") + + api_snapshot = reply.instance_snapshot + return cls._from_instance_api_snapshot(api_snapshot, api_token, host) + + except Exception as e: + raise SandboxError(f"Failed to get snapshot: {e}") from e + + @classmethod + def list( + cls, + service_id: Optional[str] = None, + snapshot_type: Optional[SnapshotType] = None, + status: Optional[SnapshotStatus] = None, + limit: int = 50, + offset: int = 0, + api_token: Optional[str] = None, + host: Optional[str] = None, + ) -> List[Snapshot]: + """ + List snapshots with optional filters. + + Uses the InstanceSnapshots API which is for sandbox/service instance snapshots. + + Args: + service_id: Filter by service ID + snapshot_type: Filter by snapshot type + status: Filter by snapshot status + limit: Maximum number of snapshots to return + offset: Offset for pagination + api_token: Koyeb API token + host: Koyeb API host + + Returns: + List of Snapshot objects + """ + import os + + if not api_token: + api_token = os.getenv("KOYEB_API_TOKEN") + if not api_token: + raise SandboxError("API token is required. Set KOYEB_API_TOKEN environment variable.") + + try: + from koyeb.api.models.instance_snapshot_type import InstanceSnapshotType + from koyeb.api.models.instance_snapshot_status import InstanceSnapshotStatus + + clients = get_api_clients(api_token, host) + + # Map our SnapshotType to InstanceSnapshotType + type_param = None + if snapshot_type: + if snapshot_type == SnapshotType.FILESYSTEM: + type_param = InstanceSnapshotType.INSTANCE_SNAPSHOT_TYPE_FILESYSTEM + elif snapshot_type == SnapshotType.FULL: + type_param = InstanceSnapshotType.INSTANCE_SNAPSHOT_TYPE_FULL + + # Map our SnapshotStatus to InstanceSnapshotStatus + status_param = None + if status: + status_map = { + SnapshotStatus.CREATING: InstanceSnapshotStatus.INSTANCE_SNAPSHOT_STATUS_CREATING, + SnapshotStatus.UPLOADING: InstanceSnapshotStatus.INSTANCE_SNAPSHOT_STATUS_UPLOADING, + SnapshotStatus.AVAILABLE: InstanceSnapshotStatus.INSTANCE_SNAPSHOT_STATUS_AVAILABLE, + SnapshotStatus.FAILED: InstanceSnapshotStatus.INSTANCE_SNAPSHOT_STATUS_FAILED, + } + status_param = status_map.get(status) + + # Build kwargs, omitting None values + list_kwargs = { + "limit": str(limit), + "offset": str(offset), + } + if service_id: + list_kwargs["service_id"] = service_id + if type_param: + list_kwargs["type"] = type_param + if status_param: + list_kwargs["status"] = status_param + + reply = clients.instance_snapshots.list_instance_snapshots(**list_kwargs) + + snapshots = [] + if reply.instance_snapshots: + for api_snapshot in reply.instance_snapshots: + snapshots.append(cls._from_instance_api_snapshot(api_snapshot, api_token, host)) + + return snapshots + + except Exception as e: + raise SandboxError(f"Failed to list snapshots: {e}") from e + + @classmethod + def _from_api_snapshot( + cls, + api_snapshot: Any, + api_token: Optional[str], + host: Optional[str], + ) -> Snapshot: + """Convert API snapshot model to Snapshot object.""" + snapshot_type = SnapshotType.FILESYSTEM + if api_snapshot.type and api_snapshot.type.value: + try: + snapshot_type = SnapshotType(api_snapshot.type.value) + except ValueError: + snapshot_type = SnapshotType.FILESYSTEM + + status = SnapshotStatus.INVALID + if api_snapshot.status and api_snapshot.status.value: + try: + status = SnapshotStatus(api_snapshot.status.value) + except ValueError: + status = SnapshotStatus.INVALID + + return cls( + id=api_snapshot.id or "", + name=api_snapshot.name or "", + service_id=api_snapshot.service_id or "", + snapshot_type=snapshot_type, + status=status, + created_at=api_snapshot.created_at or datetime.utcnow(), + deployment_id=api_snapshot.deployment_id, + available_at=api_snapshot.available_at, + size=api_snapshot.size, + region=api_snapshot.region or "", + organization_id=api_snapshot.organization_id or "", + project_id=api_snapshot.project_id, + messages=api_snapshot.messages or [], + api_token=api_token, + host=host, + ) + + @classmethod + def _from_instance_api_snapshot( + cls, + api_snapshot: Any, + api_token: Optional[str], + host: Optional[str], + sandbox_secret: Optional[str] = None, + ) -> Snapshot: + """Convert InstanceSnapshot API model to Snapshot object.""" + # Map InstanceSnapshotType to our SnapshotType + snapshot_type = SnapshotType.FILESYSTEM + if api_snapshot.type and api_snapshot.type.value: + type_map = { + "INSTANCE_SNAPSHOT_TYPE_FILESYSTEM": SnapshotType.FILESYSTEM, + "INSTANCE_SNAPSHOT_TYPE_FULL": SnapshotType.FULL, + } + snapshot_type = type_map.get(api_snapshot.type.value, SnapshotType.FILESYSTEM) + + # Map InstanceSnapshotStatus to our SnapshotStatus + status = SnapshotStatus.INVALID + if api_snapshot.status and api_snapshot.status.value: + status_map = { + "INSTANCE_SNAPSHOT_STATUS_INVALID": SnapshotStatus.INVALID, + "INSTANCE_SNAPSHOT_STATUS_CREATING": SnapshotStatus.CREATING, + "INSTANCE_SNAPSHOT_STATUS_UPLOADING": SnapshotStatus.UPLOADING, + "INSTANCE_SNAPSHOT_STATUS_AVAILABLE": SnapshotStatus.AVAILABLE, + "INSTANCE_SNAPSHOT_STATUS_DELETING": SnapshotStatus.DELETING, + "INSTANCE_SNAPSHOT_STATUS_DELETED": SnapshotStatus.DELETED, + "INSTANCE_SNAPSHOT_STATUS_FAILED": SnapshotStatus.FAILED, + } + status = status_map.get(api_snapshot.status.value, SnapshotStatus.INVALID) + + # Get region from regional_deployment_id if available, otherwise use empty string + region = "" + if api_snapshot.regional_deployment_id: + # regional_deployment_id typically contains region info, e.g., "region-service-id" + # For now, we'll leave it as empty since we don't have a direct mapping + region = "" + + return cls( + id=api_snapshot.id or "", + name=api_snapshot.name or "", + service_id=api_snapshot.service_id or "", + snapshot_type=snapshot_type, + status=status, + created_at=api_snapshot.created_at or datetime.utcnow(), + deployment_id=api_snapshot.deployment_id, + available_at=api_snapshot.available_at, + region=region, + organization_id=api_snapshot.organization_id or "", + project_id=api_snapshot.project_id, + messages=api_snapshot.messages or [], + api_token=api_token, + host=host, + sandbox_secret=sandbox_secret, + ) + + def refresh(self) -> None: + """Refresh snapshot state from the API.""" + if not self.id or not self.api_token: + raise SandboxError("Cannot refresh snapshot without ID and API token") + + updated = self.get(self.id, api_token=self.api_token, host=self.host) + # Preserve sandbox_secret, api_token, host, and operations - don't overwrite from API + for field in self.__dataclass_fields__: + if field not in ("api_token", "host", "sandbox_secret", "operations"): + setattr(self, field, getattr(updated, field)) + + def wait_available( + self, + timeout: int = 600, + poll_interval: float = 5.0, + ) -> bool: + """ + Wait for snapshot to become available. + + Args: + timeout: Maximum time to wait in seconds + poll_interval: Time between status checks in seconds + + Returns: + True if snapshot became available, False if timeout + """ + import time + + start_time = time.time() + while time.time() - start_time < timeout: + self.refresh() + if self.status == SnapshotStatus.AVAILABLE: + return True + if self.status == SnapshotStatus.FAILED: + return False + time.sleep(poll_interval) + return False + + def delete(self) -> bool: + """ + Delete this snapshot. + + Uses the InstanceSnapshots API which is for sandbox/service instance snapshots. + + Returns: + True if deletion was successful + """ + if not self.id or not self.api_token: + raise SandboxError("Cannot delete snapshot without ID and API token") + + try: + clients = get_api_clients(self.api_token, self.host) + reply = clients.instance_snapshots.delete_instance_snapshot(self.id) + return True + + except Exception as e: + raise SandboxError(f"Failed to delete snapshot: {e}") from e + + def spawn( + self, + name: Optional[str] = None, + wait_ready: bool = True, + timeout: int = 300, + **create_kwargs, + ) -> Sandbox: + """ + Spawn a new sandbox from this snapshot. + + Args: + name: Name for the new sandbox + wait_ready: Whether to wait for sandbox to be ready + timeout: Timeout for sandbox creation in seconds + **create_kwargs: Additional arguments to pass to Sandbox.create() + + Returns: + Sandbox: A new sandbox instance initialized from this snapshot + """ + if not self.id: + raise SandboxError("Cannot spawn from snapshot without ID") + + create_params = { + "snapshot": self, + "wait_ready": wait_ready, + "timeout": timeout, + } + if name: + create_params["name"] = name + if self.api_token: + create_params["api_token"] = self.api_token + if self.host: + create_params["host"] = self.host + if self.sandbox_secret: + create_params["sandbox_secret"] = self.sandbox_secret + + create_params.update(create_kwargs) + return Sandbox.create(**create_params) + + +class DeclarativeSnapshot: + """ + Fluent builder for creating sandbox snapshots declaratively. + + This builder allows you to define a sandbox environment by: + - Writing files + - Copying local files/directories + - Running setup commands + - Setting environment variables + + Then builds a snapshot that can be used to spawn pre-configured sandboxes. + """ + + def __init__( + self, + name: str, + image: str, + workdir: Optional[str] = None, + api_token: Optional[str] = None, + host: Optional[str] = None, + delete_builder: bool = True, + ): + """ + Initialize the declarative snapshot builder. + + Args: + name: Name for the template/builder + image: Docker image to use for the sandbox + workdir: Working directory in the sandbox + api_token: Koyeb API token + host: Koyeb API host + delete_builder: Whether to delete the builder sandbox after creating the snapshot (default: True) + """ + import os + + if not api_token: + api_token = os.getenv("KOYEB_API_TOKEN") + if not api_token: + raise SandboxError("API token is required. Set KOYEB_API_TOKEN environment variable.") + + self._name = name + self._image = image + self._workdir = workdir + self._api_token = api_token + self._host = host + self._delete_builder = delete_builder + self._files: Dict[str, str] = {} + self._copy_ops: List[Tuple[str, str]] = [] + self._commands: List[str] = [] + self._operations: List[str] = [] + self._builder_sandbox: Optional[Sandbox] = None + + def file(self, path: str, content: str) -> DeclarativeSnapshot: + """ + Write a file to the sandbox during build. + + Args: + path: Path in the sandbox (e.g., "/workspace/requirements.txt") + content: File content as string + + Returns: + self for method chaining + """ + self._files[path] = content + return self + + def copy(self, src: str, dst: str) -> DeclarativeSnapshot: + """ + Copy a local file or directory to the sandbox during build. + + Args: + src: Local source path (file or directory) + dst: Destination path in the sandbox + + Returns: + self for method chaining + """ + self._copy_ops.append((src, dst)) + return self + + def run(self, command: str, cwd: Optional[str] = None) -> DeclarativeSnapshot: + """ + Run a command during build. + + Args: + command: Command to execute + cwd: Working directory for the command + + Returns: + self for method chaining + """ + if cwd: + self._commands.append(f"cd {cwd} && {command}") + else: + self._commands.append(command) + return self + + def build( + self, + snapshot_name: Optional[str] = None, + ) -> Snapshot: + """ + Build the snapshot by creating a temporary sandbox, + applying all configurations, and creating a snapshot. + + The builder sandbox is automatically deleted after the snapshot is created. + + Args: + snapshot_name: Name for the snapshot (defaults to builder name) + + Returns: + Snapshot: The created snapshot + """ + import os + import shutil + + name = snapshot_name or self._name + + try: + # Create builder sandbox + self._builder_sandbox = Sandbox.create( + image=self._image, + name=f"builder-{self._name}", + wait_ready=True, + api_token=self._api_token, + host=self._host, + ) + + # Apply working directory + if self._workdir: + self._operations.append(f"set_workdir: {self._workdir}") + self._builder_sandbox.filesystem.mkdir(self._workdir) + + # Write files + for path, content in self._files.items(): + self._operations.append(f"create_file: {path}") + self._builder_sandbox.filesystem.write_file(path, content) + + # Copy local files/directories + for src, dst in self._copy_ops: + self._operations.append(f"copy: {src} -> {dst}") + if os.path.isdir(src): + # Copy directory recursively + self._copy_directory(src, dst) + else: + # Copy single file + with open(src, "r") as f: + content = f.read() + self._builder_sandbox.filesystem.write_file(dst, content) + + # Run commands + for command in self._commands: + self._operations.append(f"run: {command}") + result = self._builder_sandbox.exec(command, timeout=300) + if not result.success: + raise SandboxError(f"Command failed: {command}\nstdout: {result.stdout}\nstderr: {result.stderr}") + + # Create snapshot + snapshot = self._builder_sandbox.snapshot( + name=name, + snapshot_type=SnapshotType.FILESYSTEM, + wait_available=True, + ) + + # Store operations in the snapshot + snapshot.operations = self._operations + + return snapshot + + except Exception as e: + raise SandboxError(f"Failed to build snapshot: {e}") from e + + finally: + # Delete the builder sandbox if requested + if self._delete_builder and self._builder_sandbox: + try: + self._builder_sandbox.delete() + except Exception: + pass + + def _copy_directory(self, src_dir: str, dst_dir: str) -> None: + """Recursively copy a local directory to the sandbox.""" + if not self._builder_sandbox: + return + + for root, dirs, files in os.walk(src_dir): + # Create directories + for d in dirs: + src_path = os.path.join(root, d) + rel_path = os.path.relpath(src_path, src_dir) + dst_path = os.path.join(dst_dir, rel_path) + self._builder_sandbox.filesystem.mkdir(dst_path) + + # Copy files + for f in files: + src_path = os.path.join(root, f) + rel_path = os.path.relpath(src_path, src_dir) + dst_path = os.path.join(dst_dir, rel_path) + with open(src_path, "r") as f: + content = f.read() + self._builder_sandbox.filesystem.write_file(dst_path, content) + + def get_operations(self) -> List[str]: + """Get list of operations recorded during build.""" + return self._operations diff --git a/koyeb/sandbox/utils.py b/koyeb/sandbox/utils.py index b8483989..66950358 100644 --- a/koyeb/sandbox/utils.py +++ b/koyeb/sandbox/utils.py @@ -18,8 +18,10 @@ CatalogInstancesApi, DeploymentsApi, InstancesApi, + InstanceSnapshotsApi, SecretsApi, ServicesApi, + SnapshotsApi, ) from koyeb.api.models.config_file import ConfigFile from koyeb.api.models.deployment_definition import DeploymentDefinition @@ -107,6 +109,8 @@ class ApiClients: catalog_instances: CatalogInstancesApi deployments: DeploymentsApi secrets: SecretsApi + snapshots: Any + instance_snapshots: Any _api_clients_cache: Dict[Tuple[str, str], ApiClients] = {} @@ -156,6 +160,8 @@ def get_api_clients( catalog_instances=CatalogInstancesApi(api_client), deployments=DeploymentsApi(api_client), secrets=SecretsApi(api_client), + snapshots=SnapshotsApi(api_client), + instance_snapshots=InstanceSnapshotsApi(api_client), ) _api_clients_cache[cache_key] = clients return clients @@ -170,8 +176,10 @@ def get_api_clients( CatalogInstancesApi as AsyncCatalogInstancesApi, DeploymentsApi as AsyncDeploymentsApi, InstancesApi as AsyncInstancesApi, + InstanceSnapshotsApi as AsyncInstanceSnapshotsApi, SecretsApi as AsyncSecretsApi, ServicesApi as AsyncServicesApi, + SnapshotsApi as AsyncSnapshotsApi, ) @@ -185,6 +193,8 @@ class AsyncApiClients: catalog_instances: AsyncCatalogInstancesApi deployments: AsyncDeploymentsApi secrets: AsyncSecretsApi + snapshots: Any + instance_snapshots: Any _async_api_clients_cache: Dict[Tuple[str, str], AsyncApiClients] = {} @@ -234,6 +244,8 @@ def get_async_api_clients( catalog_instances=AsyncCatalogInstancesApi(api_client), deployments=AsyncDeploymentsApi(api_client), secrets=AsyncSecretsApi(api_client), + snapshots=AsyncSnapshotsApi(api_client), + instance_snapshots=AsyncInstanceSnapshotsApi(api_client), ) _async_api_clients_cache[cache_key] = clients return clients From b8cb2db0ac5ef2bffe393b4c23c7fb9777a548bf Mon Sep 17 00:00:00 2001 From: nicoche <78445450+nicoche@users.noreply.github.com> Date: Mon, 20 Jul 2026 17:27:27 +0200 Subject: [PATCH 2/4] Clean up snapshot-related dead code and performance issues - Remove duplicate imports in __init__.py - Remove unused datetime import in sandbox.py - Remove unused _from_api_snapshot method in snapshot.py - Fix dead code in region extraction in _from_instance_api_snapshot - Eliminate code duplication in _create_sync by building deployment definition once - Improve error handling by catching SandboxError instead of broad Exception - Remove redundant condition (snapshot_type is not None when already checked) Generated by Mistral Vibe. Co-Authored-By: Mistral Vibe --- koyeb/sandbox/__init__.py | 4 -- koyeb/sandbox/sandbox.py | 93 ++++++++++++++++----------------------- koyeb/sandbox/snapshot.py | 50 ++------------------- 3 files changed, 41 insertions(+), 106 deletions(-) diff --git a/koyeb/sandbox/__init__.py b/koyeb/sandbox/__init__.py index c1dac153..f1043801 100644 --- a/koyeb/sandbox/__init__.py +++ b/koyeb/sandbox/__init__.py @@ -26,10 +26,6 @@ SandboxError, SandboxServiceError, SandboxTimeoutError, - SandboxDeploymentError, - SandboxError, - SandboxServiceError, - SandboxTimeoutError ) __all__ = [ diff --git a/koyeb/sandbox/sandbox.py b/koyeb/sandbox/sandbox.py index d325281c..b858e260 100644 --- a/koyeb/sandbox/sandbox.py +++ b/koyeb/sandbox/sandbox.py @@ -11,7 +11,6 @@ import secrets import time from dataclasses import dataclass -from datetime import datetime from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Union from koyeb.api.api.deployments_api import DeploymentsApi @@ -248,30 +247,31 @@ def create( if snapshot is not None: if isinstance(snapshot, str): # snapshot is a snapshot ID or name string - from .snapshot import Snapshot + from .snapshot import Snapshot, SnapshotType # Try to get snapshot by ID first (fast path for UUIDs) try: snapshot_obj = Snapshot.get(snapshot, api_token=api_token, host=host) actual_snapshot_id = snapshot_obj.id actual_snapshot_type = snapshot_obj.snapshot_type - except Exception: + except SandboxError: # If that fails, try to find by name try: - snapshots = Snapshot.list(api_token=api_token, host=host, limit=100) + snapshots = Snapshot.list( + api_token=api_token, host=host, limit=100 + ) for s in snapshots: if s.name == snapshot: actual_snapshot_id = s.id actual_snapshot_type = s.snapshot_type break - except Exception: + except SandboxError: pass # If we couldn't resolve it, use the string as-is # Default to FILESYSTEM type if we couldn't determine it if actual_snapshot_id is None: actual_snapshot_id = snapshot - from .snapshot import SnapshotType actual_snapshot_type = SnapshotType.FILESYSTEM else: # snapshot is a Snapshot object @@ -419,6 +419,30 @@ def _create_sync( delete_after_sleep=delete_after_inactivity_delay, ) + # Build deployment definition - used for both snapshot and non-snapshot cases + env_vars = build_env_vars(env) + config_file_objects = build_config_files(config_files) + docker_source = create_docker_source( + image, privileged=privileged, image_registry_secret=registry_secret, + entrypoint=entrypoint, command=command, args=args, + ) + deployment_definition = create_deployment_definition( + name=name, + docker_source=docker_source, + env_vars=env_vars, + instance_type=instance_type, + exposed_port_protocol=exposed_port_protocol, + region=region, + routes=routes, + idle_timeout=idle_timeout, + enable_tcp_proxy=enable_tcp_proxy, + _experimental_enable_light_sleep=_experimental_enable_light_sleep, + _experimental_deep_sleep_value=_experimental_deep_sleep_value, + enable_mesh=enable_mesh, + config_files=config_file_objects if config_file_objects else None, + network_policy=network_policy, + ) + # Handle snapshot creation based on snapshot type # For FULL snapshots, don't provide definition (API will infer it) # For FILESYSTEM snapshots, always provide definition with snapshot_id @@ -427,7 +451,7 @@ def _create_sync( from .snapshot import SnapshotType as ST # For FULL snapshots, create service without definition - if snapshot_type == ST.FULL and snapshot_type is not None: + if snapshot_type == ST.FULL: create_service = CreateService( app_id=app_id, life_cycle=service_life_cycle, @@ -436,28 +460,6 @@ def _create_sync( ) else: # For FILESYSTEM snapshots (or unknown), provide definition - env_vars = build_env_vars(env) - config_file_objects = build_config_files(config_files) - docker_source = create_docker_source( - image, privileged=privileged, image_registry_secret=registry_secret, - entrypoint=entrypoint, command=command, args=args, - ) - deployment_definition = create_deployment_definition( - name=name, - docker_source=docker_source, - env_vars=env_vars, - instance_type=instance_type, - exposed_port_protocol=exposed_port_protocol, - region=region, - routes=routes, - idle_timeout=idle_timeout, - enable_tcp_proxy=enable_tcp_proxy, - _experimental_enable_light_sleep=_experimental_enable_light_sleep, - _experimental_deep_sleep_value=_experimental_deep_sleep_value, - enable_mesh=enable_mesh, - config_files=config_file_objects if config_file_objects else None, - network_policy=network_policy, - ) create_service = CreateService( app_id=app_id, definition=deployment_definition, @@ -467,28 +469,6 @@ def _create_sync( ) else: # No snapshot, create normally with definition - env_vars = build_env_vars(env) - config_file_objects = build_config_files(config_files) - docker_source = create_docker_source( - image, privileged=privileged, image_registry_secret=registry_secret, - entrypoint=entrypoint, command=command, args=args, - ) - deployment_definition = create_deployment_definition( - name=name, - docker_source=docker_source, - env_vars=env_vars, - instance_type=instance_type, - exposed_port_protocol=exposed_port_protocol, - region=region, - routes=routes, - idle_timeout=idle_timeout, - enable_tcp_proxy=enable_tcp_proxy, - _experimental_enable_light_sleep=_experimental_enable_light_sleep, - _experimental_deep_sleep_value=_experimental_deep_sleep_value, - enable_mesh=enable_mesh, - config_files=config_file_objects if config_file_objects else None, - network_policy=network_policy, - ) create_service = CreateService( app_id=app_id, definition=deployment_definition, @@ -1753,30 +1733,31 @@ async def create( if snapshot is not None: if isinstance(snapshot, str): # snapshot is a snapshot ID or name string - from .snapshot import Snapshot + from .snapshot import Snapshot, SnapshotType # Try to get snapshot by ID first (fast path for UUIDs) try: snapshot_obj = Snapshot.get(snapshot, api_token=api_token, host=host) actual_snapshot_id = snapshot_obj.id actual_snapshot_type = snapshot_obj.snapshot_type - except Exception: + except SandboxError: # If that fails, try to find by name try: - snapshots = Snapshot.list(api_token=api_token, host=host, limit=100) + snapshots = Snapshot.list( + api_token=api_token, host=host, limit=100 + ) for s in snapshots: if s.name == snapshot: actual_snapshot_id = s.id actual_snapshot_type = s.snapshot_type break - except Exception: + except SandboxError: pass # If we couldn't resolve it, use the string as-is # Default to FILESYSTEM type if we couldn't determine it if actual_snapshot_id is None: actual_snapshot_id = snapshot - from .snapshot import SnapshotType actual_snapshot_type = SnapshotType.FILESYSTEM else: # snapshot is a Snapshot object diff --git a/koyeb/sandbox/snapshot.py b/koyeb/sandbox/snapshot.py index 00160c5c..90a2a25c 100644 --- a/koyeb/sandbox/snapshot.py +++ b/koyeb/sandbox/snapshot.py @@ -190,46 +190,6 @@ def list( except Exception as e: raise SandboxError(f"Failed to list snapshots: {e}") from e - @classmethod - def _from_api_snapshot( - cls, - api_snapshot: Any, - api_token: Optional[str], - host: Optional[str], - ) -> Snapshot: - """Convert API snapshot model to Snapshot object.""" - snapshot_type = SnapshotType.FILESYSTEM - if api_snapshot.type and api_snapshot.type.value: - try: - snapshot_type = SnapshotType(api_snapshot.type.value) - except ValueError: - snapshot_type = SnapshotType.FILESYSTEM - - status = SnapshotStatus.INVALID - if api_snapshot.status and api_snapshot.status.value: - try: - status = SnapshotStatus(api_snapshot.status.value) - except ValueError: - status = SnapshotStatus.INVALID - - return cls( - id=api_snapshot.id or "", - name=api_snapshot.name or "", - service_id=api_snapshot.service_id or "", - snapshot_type=snapshot_type, - status=status, - created_at=api_snapshot.created_at or datetime.utcnow(), - deployment_id=api_snapshot.deployment_id, - available_at=api_snapshot.available_at, - size=api_snapshot.size, - region=api_snapshot.region or "", - organization_id=api_snapshot.organization_id or "", - project_id=api_snapshot.project_id, - messages=api_snapshot.messages or [], - api_token=api_token, - host=host, - ) - @classmethod def _from_instance_api_snapshot( cls, @@ -262,12 +222,10 @@ def _from_instance_api_snapshot( } status = status_map.get(api_snapshot.status.value, SnapshotStatus.INVALID) - # Get region from regional_deployment_id if available, otherwise use empty string - region = "" - if api_snapshot.regional_deployment_id: - # regional_deployment_id typically contains region info, e.g., "region-service-id" - # For now, we'll leave it as empty since we don't have a direct mapping - region = "" + # Get region from regional_deployment_id if available + # Note: regional_deployment_id typically contains region info, e.g., "region-service-id" + # but we don't have a direct mapping, so we use the region field directly + region = api_snapshot.region or "" return cls( id=api_snapshot.id or "", From 3f0ee40082094bea6c60a4a0dba2658238c03053 Mon Sep 17 00:00:00 2001 From: nicoche <78445450+nicoche@users.noreply.github.com> Date: Mon, 20 Jul 2026 17:37:10 +0200 Subject: [PATCH 3/4] Fix region attribute access in snapshot conversion The InstanceSnapshot API object may not have a 'region' attribute. Use hasattr() to safely check for region and regional_deployment_id. Generated by Mistral Vibe. Co-Authored-By: Mistral Vibe --- koyeb/sandbox/snapshot.py | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/koyeb/sandbox/snapshot.py b/koyeb/sandbox/snapshot.py index 90a2a25c..c2395936 100644 --- a/koyeb/sandbox/snapshot.py +++ b/koyeb/sandbox/snapshot.py @@ -222,10 +222,15 @@ def _from_instance_api_snapshot( } status = status_map.get(api_snapshot.status.value, SnapshotStatus.INVALID) - # Get region from regional_deployment_id if available - # Note: regional_deployment_id typically contains region info, e.g., "region-service-id" - # but we don't have a direct mapping, so we use the region field directly - region = api_snapshot.region or "" + # Get region from the snapshot + # Try region field first, then regional_deployment_id if available + region = "" + if hasattr(api_snapshot, 'region') and api_snapshot.region: + region = api_snapshot.region + elif hasattr(api_snapshot, 'regional_deployment_id') and api_snapshot.regional_deployment_id: + # regional_deployment_id typically contains region info, e.g., "region-service-id" + # For now, we don't have a direct mapping, so leave as empty + region = "" return cls( id=api_snapshot.id or "", From 941ec48b2176c26c68301196709da95fffe22ba0 Mon Sep 17 00:00:00 2001 From: nicoche <78445450+nicoche@users.noreply.github.com> Date: Mon, 20 Jul 2026 18:04:06 +0200 Subject: [PATCH 4/4] Remove unused SnapshotsApi from utils.py SnapshotsApi was imported and created but never used in the codebase. Only InstanceSnapshotsApi is actually used for snapshot operations. Generated by Mistral Vibe. Co-Authored-By: Mistral Vibe --- koyeb/sandbox/utils.py | 6 ------ 1 file changed, 6 deletions(-) diff --git a/koyeb/sandbox/utils.py b/koyeb/sandbox/utils.py index 66950358..525d126f 100644 --- a/koyeb/sandbox/utils.py +++ b/koyeb/sandbox/utils.py @@ -21,7 +21,6 @@ InstanceSnapshotsApi, SecretsApi, ServicesApi, - SnapshotsApi, ) from koyeb.api.models.config_file import ConfigFile from koyeb.api.models.deployment_definition import DeploymentDefinition @@ -109,7 +108,6 @@ class ApiClients: catalog_instances: CatalogInstancesApi deployments: DeploymentsApi secrets: SecretsApi - snapshots: Any instance_snapshots: Any @@ -160,7 +158,6 @@ def get_api_clients( catalog_instances=CatalogInstancesApi(api_client), deployments=DeploymentsApi(api_client), secrets=SecretsApi(api_client), - snapshots=SnapshotsApi(api_client), instance_snapshots=InstanceSnapshotsApi(api_client), ) _api_clients_cache[cache_key] = clients @@ -179,7 +176,6 @@ def get_api_clients( InstanceSnapshotsApi as AsyncInstanceSnapshotsApi, SecretsApi as AsyncSecretsApi, ServicesApi as AsyncServicesApi, - SnapshotsApi as AsyncSnapshotsApi, ) @@ -193,7 +189,6 @@ class AsyncApiClients: catalog_instances: AsyncCatalogInstancesApi deployments: AsyncDeploymentsApi secrets: AsyncSecretsApi - snapshots: Any instance_snapshots: Any @@ -244,7 +239,6 @@ def get_async_api_clients( catalog_instances=AsyncCatalogInstancesApi(api_client), deployments=AsyncDeploymentsApi(api_client), secrets=AsyncSecretsApi(api_client), - snapshots=AsyncSnapshotsApi(api_client), instance_snapshots=AsyncInstanceSnapshotsApi(api_client), ) _async_api_clients_cache[cache_key] = clients