import logging
from typing import TYPE_CHECKING
from lbox.exceptions import ResourceNotFoundError
from labelbox.orm import query
from labelbox.orm.db_object import DbObject
from labelbox.orm.model import Entity, Field, Relationship
if TYPE_CHECKING:
from labelbox import Project
logger = logging.getLogger(__name__)
[docs]class Batch(DbObject):
"""A Batch is a group of data rows submitted to a project for labeling
Attributes:
name (str)
created_at (datetime)
updated_at (datetime)
deleted (bool)
project (Relationship): `ToOne` relationship to Project
created_by (Relationship): `ToOne` relationship to User
"""
name = Field.String("name")
created_at = Field.DateTime("created_at")
updated_at = Field.DateTime("updated_at")
size = Field.Int("size")
consensus_settings = Field.Json("consensus_settings_json")
# Relationships
created_by = Relationship.ToOne("User")
def __init__(
self, client, project_id, *args, failed_data_row_ids=[], **kwargs
):
super().__init__(client, *args, **kwargs)
self.project_id = project_id
self._failed_data_row_ids = failed_data_row_ids
[docs] def project(self) -> "Project": # type: ignore
"""Returns Project which this Batch belongs to
Raises:
LabelboxError: if the project is not found
"""
query_str = """query getProjectPyApi($projectId: ID!) {
project(
where: {id: $projectId}){
%s
}}""" % query.results_query_part(Entity.Project)
params = {"projectId": self.project_id}
response = self.client.execute(query_str, params)
if response is None:
raise ResourceNotFoundError(Entity.Project, params)
return Entity.Project(self.client, response["project"])
[docs] def remove_queued_data_rows(self) -> None:
"""Removes remaining queued data rows from the batch and labeling queue.
Args:
batch (Batch): Batch to remove queued data rows from
"""
project_id_param = "projectId"
batch_id_param = "batchId"
self.client.execute(
"""mutation RemoveQueuedDataRowsFromBatchPyApi($%s: ID!, $%s: ID!) {
project(where: {id: $%s}) { removeQueuedDataRowsFromBatch(batchId: $%s) { id } }
}"""
% (
project_id_param,
batch_id_param,
project_id_param,
batch_id_param,
),
{project_id_param: self.project_id, batch_id_param: self.uid},
experimental=True,
)
[docs] def delete(self) -> None:
"""Deletes the given batch.
Note: Batch deletion for batches that has labels is forbidden.
Args:
batch (Batch): Batch to remove queued data rows from
"""
project_id_param = "projectId"
batch_id_param = "batchId"
self.client.execute(
"""mutation DeleteBatchPyApi($%s: ID!, $%s: ID!) {
project(where: {id: $%s}) { deleteBatch(batchId: $%s) { deletedBatchId } }
}"""
% (
project_id_param,
batch_id_param,
project_id_param,
batch_id_param,
),
{project_id_param: self.project_id, batch_id_param: self.uid},
experimental=True,
)
[docs] def delete_labels(self, set_labels_as_template=False) -> None:
"""Deletes labels that were created for data rows in the batch.
Args:
batch (Batch): Batch to remove queued data rows from
set_labels_as_template (bool): When set to true, the deleted labels will be kept as templates.
"""
project_id_param = "projectId"
batch_id_param = "batchId"
type_param = "type"
res = self.client.execute(
"""mutation DeleteBatchLabelsPyApi($%s: ID!, $%s: ID!, $%s: DeleteBatchLabelsType!) {
project(where: {id: $%s}) { deleteBatchLabels(batchId: $%s, data:{ type: $%s }) { deletedLabelIds } }
}"""
% (
project_id_param,
batch_id_param,
type_param,
project_id_param,
batch_id_param,
type_param,
),
{
project_id_param: self.project_id,
batch_id_param: self.uid,
type_param: "RequeueDataWithLabelAsTemplate"
if set_labels_as_template
else "RequeueData",
},
experimental=True,
)
return res
# modify this function to return an empty list if there are no failed data rows
@property
def failed_data_row_ids(self):
if self._failed_data_row_ids is None:
self._failed_data_row_ids = []
return (x for x in self._failed_data_row_ids)