boxlite/examples/python/08_rest_api/clone_export_import.py at main · silverzki/boxlite · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
#!/usr/bin/env python3
"""
Clone, Export, and Import via REST API

Demonstrates:
- clone_box(): Clone a box on a remote server
- export(): Download a .boxlite archive from the server
- import_box(): Upload an archive to create a new box

Prerequisites:
    make dev:python
    boxlite serve --port 8100
"""

import asyncio
import tempfile

from boxlite import ApiKeyCredential, Boxlite, BoxOptions, BoxliteRestOptions


SERVER_URL = "http://localhost:8100"


def connect() -> Boxlite:
    return Boxlite.rest(BoxliteRestOptions(
        url=SERVER_URL, credential=ApiKeyCredential("local-dev-key"),
    ))


async def test_clone():
    """Clone a box via REST."""
    print("\n=== Clone via REST ===")

    rt = connect()

    # Create source box
    source = await rt.create(
        BoxOptions(image="alpine:latest", auto_remove=False),
        name="rest-clone-source",
    )
    print(f"  Source: {source.id}")

    # Clone it
    cloned = await source.clone_box(name="rest-clone-target")
    print(f"  Cloned: {cloned.id}")

    # Verify different IDs
    assert source.id != cloned.id, "Clone should have different ID"
    cloned_info = await cloned.info()
    print(f"  Clone name: {cloned_info.name}")
    print(f"  Clone state: {cloned_info.state}")

    # Cleanup
    await rt.remove(cloned.id, force=True)
    await rt.remove(source.id, force=True)
    print("  Cleaned up")


async def test_export_import():
    """Export and import a box via REST."""
    print("\n\n=== Export/Import via REST ===")

    rt = connect()

    # Create and start source box
    source = await rt.create(
        BoxOptions(image="alpine:latest", auto_remove=False),
        name="rest-export-source",
    )
    print(f"  Source: {source.id}")

    # Export to a temp directory
    with tempfile.TemporaryDirectory() as export_dir:
        print(f"  Exporting to {export_dir}...")
        archive_path = await source.export(dest=export_dir)
        print(f"  Archive: {archive_path}")

        # Import the archive
        print("  Importing archive...")
        imported = await rt.import_box(archive_path, name="rest-imported")
        print(f"  Imported: {imported.id}")

        imported_info = await imported.info()
        print(f"  Import name: {imported_info.name}")
        print(f"  Import state: {imported_info.state}")

        # Cleanup
        await rt.remove(imported.id, force=True)

    await rt.remove(source.id, force=True)
    print("  Cleaned up")


async def main():
    print("=" * 50)
    print("REST API: Clone, Export, and Import")
    print("=" * 50)

    await test_clone()
    await test_export_import()

    print("\n  Done")


if __name__ == "__main__":
    asyncio.run(main())