Add sync_external_project

Author: lb-kamilk

libs/labelbox/src/labelbox/__init__.py

@@ -22,6 +22,15 @@
 )
 from labelbox.schema.dataset import Dataset
 from labelbox.schema.enums import AnnotationImportState
+from labelbox.schema.project_sync import (
+    AutoQA,
+    AutoQaStatus,
+    CustomScore,
+    ProjectSyncEntry,
+    ProjectSyncLabel,
+    ProjectSyncResult,
+    SubmittedBy,
+)
 from labelbox.schema.export_task import (
     BufferedJsonConverterOutput,
     ExportTask,

libs/labelbox/src/labelbox/schema/external_metrics.py

@@ -0,0 +1,16 @@
+"""Backwards-compatibility re-exports. Prefer importing from project_sync."""
+
+from labelbox.schema.project_sync import (  # noqa: F401
+    AutoQA,
+    AutoQaStatus,
+    CustomScore,
+    ProjectSyncEntry,
+    ProjectSyncLabel,
+    ProjectSyncResult,
+    SubmittedBy,
+    _to_gql_input,
+)
+
+# Deprecated aliases — kept so existing callers don't break immediately.
+ExternalMetricsEntry = ProjectSyncEntry
+ExternalMetricsResult = ProjectSyncResult

libs/labelbox/src/labelbox/schema/project.py

@@ -37,6 +37,11 @@
     ProjectExportFilters,
     build_filters,
 )
+from labelbox.schema.project_sync import (
+    ProjectSyncEntry,
+    ProjectSyncResult,
+    _to_gql_input,
+)
 from labelbox.schema.export_params import ProjectExportParams
 from labelbox.schema.export_task import ExportTask
 from labelbox.schema.identifiable import DataRowIdentifier
@@ -1001,6 +1006,39 @@ def create_batches(
 
         return CreateBatchesTask(self.client, self.uid, batch_ids, task_ids)
 
+    def sync_external_project(
+        self,
+        entries: List[ProjectSyncEntry],
+    ) -> ProjectSyncResult:
+        """Syncs external project data — labels, metrics, and workflow state.
+
+        Processing is asynchronous. The returned submission ID can be used
+        to track the progress of the sync operation.
+
+        Args:
+            entries: A list of ProjectSyncEntry objects.
+
+        Returns:
+            A ProjectSyncResult containing the submission ID.
+        """
+        mutation_str = """mutation syncExternalProjectPyApi($input: SyncExternalProjectInput!) {
+            syncExternalProject(input: $input) {
+                submissionId
+            }
+        }"""
+
+        params = {
+            "input": {
+                "projectId": self.uid,
+                "entries": [_to_gql_input(e) for e in entries],
+            }
+        }
+
+        response = self.client.execute(mutation_str, params)
+        payload = response["syncExternalProject"]
+
+        return ProjectSyncResult(submission_id=payload["submissionId"])
+
     def create_batches_from_dataset(
         self,
         name_prefix: str,

libs/labelbox/src/labelbox/schema/project_sync.py

@@ -0,0 +1,85 @@
+from enum import Enum
+from typing import Any, Dict, List, Optional
+
+from pydantic import BaseModel
+
+
+class AutoQaStatus(str, Enum):
+    Approve = "Approve"
+    Reject = "Reject"
+    Neutral = "Neutral"
+
+
+class SubmittedBy(BaseModel):
+    email: str
+
+
+class CustomScore(BaseModel):
+    name: str
+    value: float
+
+
+class AutoQA(BaseModel):
+    status: AutoQaStatus
+    score: Optional[float] = None
+    feedback: Optional[str] = None
+    custom_scores: Optional[List[CustomScore]] = None
+
+
+class ProjectSyncLabel(BaseModel):
+    submitted_by: SubmittedBy
+    auto_qa: Optional[AutoQA] = None
+    seconds_to_completion: Optional[float] = None
+    submitted_on: Optional[str] = None
+
+
+class ProjectSyncEntry(BaseModel):
+    task_id: str
+    label: Optional[ProjectSyncLabel] = None
+    queue_type: Optional[str] = None
+    row_status: Optional[str] = None
+
+
+class ProjectSyncResult(BaseModel):
+    submission_id: str
+
+
+def _to_gql_input(entry: ProjectSyncEntry) -> Dict[str, Any]:
+    """Convert a ProjectSyncEntry to a camelCase dict matching the GQL schema."""
+    result: Dict[str, Any] = {"taskId": entry.task_id}
+
+    if entry.label is not None:
+        label: Dict[str, Any] = {
+            "submittedBy": {"email": entry.label.submitted_by.email},
+        }
+
+        if entry.label.auto_qa is not None:
+            auto_qa: Dict[str, Any] = {
+                "status": entry.label.auto_qa.status.value,
+            }
+            if entry.label.auto_qa.score is not None:
+                auto_qa["score"] = entry.label.auto_qa.score
+            if entry.label.auto_qa.feedback is not None:
+                auto_qa["feedback"] = entry.label.auto_qa.feedback
+            if entry.label.auto_qa.custom_scores is not None:
+                auto_qa["customScores"] = [
+                    {"name": cs.name, "value": cs.value}
+                    for cs in entry.label.auto_qa.custom_scores
+                ]
+            label["autoQA"] = auto_qa
+
+        if entry.label.seconds_to_completion is not None:
+            label["secondsToCompletion"] = entry.label.seconds_to_completion
+
+        if entry.label.submitted_on is not None:
+            label["submittedOn"] = entry.label.submitted_on
+
+        result["label"] = label
+
+    if entry.queue_type is not None:
+        result["queueType"] = entry.queue_type
+
+    if entry.row_status is not None:
+        result["rowStatus"] = entry.row_status
+
+    return result