"""This file and its contents are licensed under the Apache License 2.0. Please see the included NOTICE for copyright information and LICENSE for a copy of the license.
"""
from enum import Enum
from typing import Any, List, Optional, Union
from pydantic import BaseModel, StrictBool, StrictFloat, StrictInt, StrictStr
class FilterIn(BaseModel):
min: Union[StrictInt, StrictFloat, StrictStr]
max: Union[StrictInt, StrictFloat, StrictStr]
class Filter(BaseModel):
child_filter: Optional['Filter'] = None
filter: str
operator: str
type: str
value: Union[StrictInt, StrictFloat, StrictBool, StrictStr, FilterIn, list]
class ConjunctionEnum(Enum):
OR = 'or'
AND = 'and'
class Filters(BaseModel):
conjunction: ConjunctionEnum
items: List[Filter]
class SelectedItems(BaseModel):
all: bool
included: List[int] = []
excluded: List[int] = []
class PrepareParams(BaseModel):
project: Union[int, List[int]] # Support both single project and multiple projects
ordering: List[str] = []
selectedItems: Optional[SelectedItems] = None
filters: Optional[Filters] = None
data: Optional[dict] = None
request: Optional[Any] = None
@property
def projects(self) -> List[int]:
"""Get project IDs as a list, whether single or multiple were provided."""
if isinstance(self.project, list):
return self.project
return [self.project]
@property
def is_multi_project(self) -> bool:
"""Check if this PrepareParams includes multiple projects."""
return isinstance(self.project, list) and len(self.project) > 1
class CustomEnum(Enum):
def __init__(self, value, description):
self._value_ = value
self.description = description
@classmethod
def enums(cls):
return sorted([item.value for item in cls])
@classmethod
def descriptions(cls):
return {item.value: item.description for item in sorted(cls, key=lambda x: x.value)}
class Column(Enum):
ID = 'id', 'Number', 'Task ID'
INNER_ID = 'inner_id', 'Number', 'Task Inner ID, it starts from 1 for all projects'
GROUND_TRUTH = 'ground_truth', 'Boolean', 'Ground truth status of the tasks'
ANNOTATIONS_RESULTS = 'annotations_results', 'String', 'Annotation results for the tasks'
REVIEWED = 'reviewed', 'Boolean', 'Whether the tasks have been reviewed (Enterprise only)'
PREDICTIONS_SCORE = 'predictions_score', 'Number', 'Prediction score for the task'
PREDICTIONS_MODEL_VERSIONS = 'predictions_model_versions', 'String', 'Model version used for the predictions'
PREDICTIONS_RESULTS = 'predictions_results', 'String', 'Prediction results for the tasks'
FILE_UPLOAD = 'file_upload', 'String', 'Name of the file uploaded to create the tasks'
CREATED_AT = 'created_at', 'Datetime', 'Time the task was created at'
UPDATED_AT = (
'updated_at',
'Datetime',
'Time the task was updated at (e.g. new annotation was created, review added, etc)',
)
ANNOTATORS = (
'annotators',
'List',
'Annotators that completed the task (Community). Can include assigned annotators (Enterprise only). '
'Important note: the filter `type` should be List, but the filter `value` is integer',
)
TOTAL_PREDICTIONS = 'total_predictions', 'Number', 'Total number of predictions for the task'
CANCELLED_ANNOTATIONS = (
'cancelled_annotations',
'Number',
'Number of cancelled or skipped annotations for the task',
)
TOTAL_ANNOTATIONS = 'total_annotations', 'Number', 'Total number of annotations on a task'
COMPLETED_AT = 'completed_at', 'Datetime', 'Time when a task was fully annotated'
AGREEMENT = 'agreement', 'Number', 'Agreement for annotation results for a specific task (Enterprise only)'
REVIEWERS = (
'reviewers',
'String',
'Reviewers that reviewed the task, or assigned reviewers (Enterprise only). '
'Important note: the filter `type` should be List, but the filter `value` is integer',
)
REVIEWS_REJECTED = (
'reviews_rejected',
'Number',
'Number of annotations rejected for a task in review (Enterprise only)',
)
REVIEWS_ACCEPTED = (
'reviews_accepted',
'Number',
'Number of annotations accepted for a task in review (Enterprise only)',
)
COMMENTS = 'comments', 'Number', 'Number of comments in a task'
UNRESOLVED_COMMENT_COUNT = 'unresolved_comment_count', 'Number', 'Number of unresolved comments in a task'
def __init__(self, value, value_type, description):
self._value_ = value
self.type = value_type
self.description = description
@classmethod
def enums_for_filters(cls):
return sorted(['filter:tasks:' + str(item.value) for item in cls])
@classmethod
def enums_for_ordering(cls):
return sorted([('tasks:' + str(item.value)) for item in cls])
@classmethod
def descriptions_for_filters(cls):
return {
'filter:tasks:' + item.value: f'({item.type}) ' + item.description
for item in sorted(cls, key=lambda x: x.value)
}
class Operator(CustomEnum):
EQUAL = 'equal', 'Equal to'
NOT_EQUAL = 'not_equal', 'Not equal to'
GREATER = 'greater', 'Greater than'
GREATER_OR_EQUAL = 'greater_or_equal', 'Greater than or equal to'
LESS = 'less', 'Less than'
LESS_OR_EQUAL = 'less_or_equal', 'Less than or equal to'
CONTAINS = 'contains', 'Contains'
NOT_CONTAINS = 'not_contains', 'Does not contain'
EXISTS = 'exists', 'Exists'
NOT_EXISTS = 'not_exists', 'Does not exist'
STARTS_WITH = 'starts_with', 'Starts with'
ENDS_WITH = 'ends_with', 'Ends with'
IS_BETWEEN = 'in', 'Is between min and max values, so the filter `value` should be e.g. `{"min": 1, "max": 7}`'
NOT_BETWEEN = (
'not_in',
'Is not between min and max values, so the filter `value` should be e.g. `{"min": 1, "max": 7}`',
)
class Type(CustomEnum):
Number = 'Number', 'Float or Integer'
Datetime = 'Datetime', "Datetime string in `strftime('%Y-%m-%dT%H:%M:%S.%fZ')` format"
Boolean = 'Boolean', 'Boolean'
String = 'String', 'String'
List = 'List', 'List of items'
Unknown = 'Unknown', 'Unknown is explicitly converted to string format'
# Example request and response
example_request_1 = {
'filters': {
'conjunction': 'or',
'items': [{'filter': 'filter:tasks:id', 'operator': 'greater', 'type': 'Number', 'value': 123}],
},
'selectedItems': {'all': True, 'excluded': [124, 125, 126]},
'ordering': ['tasks:total_annotations'],
}
example_request_2 = {
'filters': {
'conjunction': 'or',
'items': [
{
'filter': 'filter:tasks:completed_at',
'operator': 'in',
'type': 'Datetime',
'value': {'min': '2021-01-01T00:00:00.000Z', 'max': '2025-01-01T00:00:00.000Z'},
}
],
},
'selectedItems': {'all': False, 'included': [1, 2, 3]},
'ordering': ['-tasks:completed_at'],
}
# Define the schemas for filters and selectedItems
filters_schema = {
'type': 'object',
'properties': {
'conjunction': {
'type': 'string',
'enum': ['or', 'and'],
'description': (
'Logical conjunction for the filters. This conjunction (either "or" or "and") '
'will be applied to all items in the filters list. It is not possible to combine '
'"or" and "and" within one list of filters. All filters will be either combined with "or" '
'or with "and", but not a mix of both.'
),
},
'items': {
'type': 'array',
'items': {
'type': 'object',
'properties': {
'filter': {
'type': 'string',
'enum': Column.enums_for_filters(),
'description': (
'Filter identifier, it should start with `filter:tasks:` prefix, '
'e.g. `filter:tasks:agreement`. '
'For `task.data` fields it may look like `filter:tasks:data.field_name`. '
'If you need more info about columns, check the '
'[Get data manager columns](#tag/Data-Manager/operation/api_dm_columns_list) API endpoint. '
'Possible values:
'
+ '
'.join(
[
f'