Export and Convert Snapshots

1
# Define the URL where Label Studio is accessible
2
LABEL_STUDIO_URL = 'YOUR_BASE_URL'
3
# API key is available at the Account & Settings page in Label Studio UI
4
LABEL_STUDIO_API_KEY = 'YOUR_ACCESS_TOKEN'
5
PROJECT_ID=1    # replace with your project ID 
6
7
import sys
8
import time
9
from pathlib import Path
10
import logging
11
12
from label_studio_sdk import LabelStudio
13
from label_studio_sdk.core.api_error import ApiError
14
15
logging.basicConfig(level=logging.INFO)
16
logger = logging.getLogger("export_snapshots")
17
18
19
def export_and_download_snapshot():
20
    """Export project JSON snapshot
21
22
    This example demonstrates how to:
23
    - Create an export snapshot
24
    - Poll for completion
25
    - Download the resulting file
26
    """
27
28
    # Initialize v2 client
29
    client = LabelStudio(base_url=LABEL_STUDIO_URL, api_key=LABEL_STUDIO_API_KEY)
30
31
    # Fetch project and optional first view id
32
    client.projects.get(id=PROJECT_ID)
33
    views = client.views.list(project=PROJECT_ID)
34
    {"view": views[0].id} if views else None
35
36
    # Create export snapshot
37
    create_kwargs = {
38
        "title": "Export SDK Snapshot",
39
        # task_filter_options follows API schema; pass only if a view is available
40
        # "task_filter_options": {"view": task_filter_options["view"]} if task_filter_options else None,
41
    }
42
    # Remove None keys to avoid sending them
43
    create_kwargs = {k: v for k, v in create_kwargs.items() if v is not None}
44
45
    export_job = client.projects.exports.create(id=PROJECT_ID, **create_kwargs)
46
    export_id = export_job.id
47
    logger.info(f"Created export snapshot: id={export_id}, status={export_job.status}")
48
49
    # Poll until completed or failed
50
    start = time.time()
51
    timeout_sec = 300
52
    logger.info("Waiting for export snapshot to complete...")
53
    while True:
54
        job = client.projects.exports.get(id=PROJECT_ID, export_pk=export_id)
55
        elapsed = int(time.time() - start)
56
        logger.info(f"Export status: {job.status} (elapsed {elapsed}s)")
57
        if job.status in ("completed", "failed"):
58
            break
59
        if time.time() - start > timeout_sec:
60
            raise TimeoutError(
61
                f"Export job timed out (id={export_id}, status={job.status})"
62
            )
63
        time.sleep(1.0)
64
65
    if job.status == "failed":
66
        raise ApiError(status_code=500, body=f"Export failed: {job}")
67
68
    # Download export as JSON to local file
69
    out_dir = Path(".")
70
    out_dir.mkdir(parents=True, exist_ok=True)
71
    out_path = out_dir / f"project_{PROJECT_ID}_export_{export_id}.json"
72
73
    with open(out_path, "wb") as f:
74
        for chunk in client.projects.exports.download(
75
            id=PROJECT_ID,
76
            export_pk=export_id,
77
            export_type="JSON",
78
            request_options={"chunk_size": 1024},
79
        ):
80
            f.write(chunk)
81
82
    logger.info(f"Export completed. File saved to: {out_path}")
83
    return export_id
84
85
86
def convert_snapshot(export_type: str, export_id: str = None):
87
    """Convert the project export JSON snapshot to a specified format and download it.
88
89
    The conversion API is used to start a conversion on the Label Studio backend, then we poll export status
90
    until the specific converted format is completed, and finally download it.
91
    See docs: projects/exports/convert and projects/exports/list.
92
93
    Args:
94
        export_type: The type of export to convert to.
95
        export_id: The ID of the export to convert. If not provided, the latest export will be used.
96
    """
97
    client = LabelStudio(base_url=LABEL_STUDIO_URL, api_key=LABEL_STUDIO_API_KEY)
98
99
    # Get the latest export snapshot for the project
100
    exports = client.projects.exports.list(id=PROJECT_ID)
101
    if not exports:
102
        raise ApiError(
103
            status_code=404, body="No export snapshots found for the project"
104
        )
105
    exports = sorted(exports, key=lambda e: e.created_at, reverse=True)
106
    export = exports[0]
107
    export_id = export.id if not export_id else export_id
108
109
    # Start conversion
110
    conv = client.projects.exports.convert(
111
        export_pk=export_id, id=PROJECT_ID, export_type=export_type
112
    )
113
    converted_format_id = conv.converted_format
114
    logger.info(
115
        f"Started conversion: export_id={export_id}, export_type={export_type}, converted_format_id={converted_format_id}"
116
    )
117
118
    # Poll converted format status
119
    start = time.time()
120
    timeout_sec = 300
121
    logger.info("Waiting for conversion to complete...")
122
    while True:
123
        cur = client.projects.exports.get(id=PROJECT_ID, export_pk=export_id)
124
        cf = None
125
        if cur.converted_formats:
126
            cf = next(
127
                (
128
                    c
129
                    for c in cur.converted_formats
130
                    if (converted_format_id and c.id == converted_format_id)
131
                    or (c.export_type == export_type)
132
                ),
133
                None,
134
            )
135
        status = getattr(cf, "status", None)
136
        elapsed = int(time.time() - start)
137
        logger.info(
138
            f"Conversion status: {status or 'pending'} (format {export_type}, elapsed {elapsed}s)"
139
        )
140
        if status in ("completed", "failed"):
141
            break
142
        if time.time() - start > timeout_sec:
143
            raise TimeoutError(
144
                f"Conversion timed out (export_id={export_id}, format={export_type}, status={status})"
145
            )
146
        time.sleep(1.0)
147
148
    if status == "failed":
149
        raise ApiError(
150
            status_code=500,
151
            body=f"Conversion failed (export_id={export_id}, format={export_type})",
152
        )
153
154
    # Download converted file using export_type param
155
    ext = "json" if export_type.upper().startswith("JSON") else export_type.lower()
156
    out_dir = Path(".")
157
    out_dir.mkdir(parents=True, exist_ok=True)
158
    out_path = out_dir / f"project_{PROJECT_ID}_export_{export_id}.{ext}"
159
160
    with open(out_path, "wb") as f:
161
        for chunk in client.projects.exports.download(
162
            id=PROJECT_ID,
163
            export_pk=export_id,
164
            export_type=export_type,
165
            request_options={"chunk_size": 1024},
166
        ):
167
            f.write(chunk)
168
169
    logger.info(f"Converted export downloaded. File saved to: {out_path}")
170
171
172
if __name__ == "__main__":
173
174
    export_id = export_and_download_snapshot()
175
    logger.info(f"Export ID: {export_id}")
176
    
177
    export_type = sys.argv[1] if len(sys.argv) > 1 else None
178
    if export_type and export_type != "JSON":
179
        logger.info(f"Converting export to {export_type} format")
180
        convert_snapshot(export_type, export_id)