bulk export queued data rows

Author: Matt Sokoloff

labelbox/schema/project.py

@@ -160,6 +160,35 @@ def labels(self, datasets=None, order_by=None):
         return PaginatedCollection(self.client, query_str, {id_param: self.uid},
                                    ["project", "labels"], Label)
 
+    def export_queued_data_rows(self, timeout_seconds=120):
+        """ Returns all data rows that are currently enqueued for this project.
+
+        Args:
+            timeout_seconds (float): Max waiting time, in seconds.
+        Returns:
+            URL of the data file with this DataRow information. If the server didn't
+            generate during the `timeout_seconds` period, None is returned.
+        """
+        id_param = "projectId"
+        query_str = """mutation GetQueuedDataRowsExportUrlPyApi($%s: ID!)
+            {exportQueuedDataRows(data:{projectId: $%s }) {downloadUrl createdAt status} }
+        """ % (id_param, id_param)
+        sleep_time = 2
+        while True:
+            res = self.client.execute(query_str, {id_param: self.uid})
+            res = res["exportQueuedDataRows"]
+            if res["status"] == "COMPLETE":
+                return res["downloadUrl"]
+
+            timeout_seconds -= sleep_time
+            if timeout_seconds <= 0:
+                return None
+
+            logger.debug(
+                "Project '%s' queued data row export, waiting for server...",
+                self.uid)
+            time.sleep(sleep_time)
+
     def export_labels(self, timeout_seconds=60):
         """ Calls the server-side Label exporting that generates a JSON
         payload, and returns the URL to that payload.
@@ -193,13 +222,13 @@ def export_labels(self, timeout_seconds=60):
             time.sleep(sleep_time)
 
     def export_issues(self, status=None):
-        """ Calls the server-side Issues exporting that 
+        """ Calls the server-side Issues exporting that
         returns the URL to that payload.
 
         Args:
             status (string): valid values: Open, Resolved
         Returns:
-            URL of the data file with this Project's issues. 
+            URL of the data file with this Project's issues.
         """
         id_param = "projectId"
         status_param = "status"
@@ -229,14 +258,14 @@ def export_issues(self, status=None):
     def upsert_instructions(self, instructions_file: str):
         """
         * Uploads instructions to the UI. Running more than once will replace the instructions
-            
+
         Args:
             instructions_file (str): Path to a local file.
                 * Must be either a pdf, text, or html file.
 
         Raises:
             ValueError:
-                * project must be setup 
+                * project must be setup
                 * instructions file must end with one of ".text", ".txt", ".pdf", ".html"
         """
 
@@ -267,18 +296,18 @@ def upsert_instructions(self, instructions_file: str):
 
         self.client.execute(
             """mutation UpdateFrontendWithExistingOptionsPyApi (
-                    $frontendId: ID!, 
-                    $optionsId: ID!, 
-                    $name: String!, 
-                    $description: String!, 
+                    $frontendId: ID!,
+                    $optionsId: ID!,
+                    $name: String!,
+                    $description: String!,
                     $customizationOptions: String!
                 ) {
                     updateLabelingFrontend(
-                        where: {id: $frontendId}, 
+                        where: {id: $frontendId},
                         data: {name: $name, description: $description}
                     ) {id}
                     updateLabelingFrontendOptions(
-                        where: {id: $optionsId}, 
+                        where: {id: $optionsId},
                         data: {customizationOptions: $customizationOptions}
                     ) {id}
                 }""", {
@@ -390,10 +419,10 @@ def validate_labeling_parameter_overrides(self, data):
 
     def set_labeling_parameter_overrides(self, data):
         """ Adds labeling parameter overrides to this project.
-                
+
         See information on priority here:
             https://docs.labelbox.com/en/configure-editor/queue-system#reservation-system
-    
+
             >>> project.set_labeling_parameter_overrides([
             >>>     (data_row_1, 2, 3), (data_row_2, 1, 4)])
 
@@ -407,11 +436,11 @@ def set_labeling_parameter_overrides(self, data):
                         - Minimum priority is 1.
                     * Priority is not the queue position.
                         - The position is determined by the relative priority.
-                        - E.g. [(data_row_1, 5,1), (data_row_2, 2,1), (data_row_3, 10,1)] 
+                        - E.g. [(data_row_1, 5,1), (data_row_2, 2,1), (data_row_3, 10,1)]
                             will be assigned in the following order: [data_row_2, data_row_1, data_row_3]
                     * Datarows with parameter overrides will appear before datarows without overrides.
                     * The priority only effects items in the queue.
-                        - Assigning a priority will not automatically add the item back into the queue.  
+                        - Assigning a priority will not automatically add the item back into the queue.
                 Number of labels:
                     * The number of times a data row should be labeled.
                         - Creates duplicate data rows in a project (one for each number of labels).
@@ -458,7 +487,7 @@ def unset_labeling_parameter_overrides(self, data_rows):
     def upsert_review_queue(self, quota_factor):
         """ Sets the the proportion of total assets in a project to review.
 
-        More information can be found here: 
+        More information can be found here:
             https://docs.labelbox.com/en/quality-assurance/review-labels#configure-review-percentage
 
         Args:

tests/integration/conftest.py

@@ -1,7 +1,6 @@
 from collections import namedtuple
 from enum import Enum
 from datetime import datetime
-from labelbox.orm.db_object import experimental
 from random import randint
 from string import ascii_letters
 from types import SimpleNamespace
@@ -14,6 +13,7 @@
 from labelbox.schema.invite import Invite
 from labelbox.pagination import PaginatedCollection
 from labelbox.schema.user import User
+from labelbox import LabelingFrontend
 from labelbox import Client
 
 IMG_URL = "https://picsum.photos/200/300"
@@ -83,7 +83,7 @@ def get_invites(client):
     Do not use. Only for testing.
     """
     query_str = """query GetOrgInvitationsPyApi($from: ID, $first: PageSize) {
-            organization { id invites(from: $from, first: $first) { 
+            organization { id invites(from: $from, first: $first) {
                 nodes { id createdAt organizationRoleName inviteeEmail } nextCursor }}}"""
     invites = PaginatedCollection(
         client,
@@ -199,13 +199,13 @@ def organization(client):
 def project_based_user(client, rand_gen):
     email = rand_gen(str)
     # Use old mutation because it doesn't require users to accept email invites
-    query_str = """mutation MakeNewUserPyApi { 
-        addMembersToOrganization( 
-            data: { 
-                emails: ["%s@labelbox.com"], 
+    query_str = """mutation MakeNewUserPyApi {
+        addMembersToOrganization(
+            data: {
+                emails: ["%s@labelbox.com"],
                 orgRoleId: "%s",
                 projectRoles: []
-            } 
+            }
         ) {
         newUserId
         }
@@ -227,3 +227,17 @@ def project_pack(client):
     yield projects
     for proj in projects:
         proj.delete()
+
+
+@pytest.fixture
+def configured_project(project, client, rand_gen):
+    dataset = client.create_dataset(name=rand_gen(str), projects=project)
+    dataset.create_data_row(row_data=IMG_URL)
+    editor = list(
+        project.client.get_labeling_frontends(
+            where=LabelingFrontend.name == "editor"))[0]
+    empty_ontology = {"tools": [], "classifications": []}
+    project.setup(editor, empty_ontology)
+    yield project
+    dataset.delete()
+    project.delete()

tests/integration/test_project.py

@@ -1,6 +1,8 @@
-import pytest
 import json
+
 import requests
+import ndjson
+import pytest
 
 from labelbox import Project, LabelingFrontend
 from labelbox.exceptions import InvalidQueryError
@@ -100,3 +102,10 @@ def test_attach_instructions(client, project):
     with pytest.raises(ValueError) as exc_info:
         project.upsert_instructions('/tmp/file.invalid_file_extension')
     assert "instructions_file must end with one of" in str(exc_info.value)
+
+
+def test_queued_data_row_export(configured_project):
+    url = configured_project.export_queued_data_rows()
+    assert url
+    result = ndjson.loads(requests.get(url).text)
+    assert len(result) == 1