Class PreparedTaskManager

label_studio/data_manager/managers.py:773–875

Manager for Task model with Data Manager annotations. Provides: - Advanced query annotations for Data Manager - Filter and ordering support - FSM state annotation support (via TaskQuerySet) Note: Overrides the base get_queryset() to return TaskQuerySet. Also has a cust

class PreparedTaskManager(models.Manager):
    """
    Manager for Task model with Data Manager annotations.

    Provides:
    - Advanced query annotations for Data Manager
    - Filter and ordering support
    - FSM state annotation support (via TaskQuerySet)

    Note: Overrides the base get_queryset() to return TaskQuerySet. Also has
    a custom get_queryset(fields_for_evaluation, prepare_params, ...) method
    for Data Manager-specific functionality.
    """

    @staticmethod
    def annotate_queryset(
        queryset, fields_for_evaluation=None, all_fields=False, excluded_fields_for_evaluation=None, request=None
    ):
        annotations_map = get_annotations_map()
        # If we have dynamic control-tag level agreement columns, inject into annotation map
        # without mutating the global map
        if flag_set('fflag_utc_428_consensus_control_tag_agreement', user='auto'):
            inject_path = getattr(settings, 'GET_DYNAMIC_DM_ANNOTATIONS', None)
            if inject_path:
                overlay_func = load_func(inject_path)
                # Expect a dict of {field_name: function that annotates the queryset}
                overlay_map = overlay_func(request=request, project=getattr(queryset.first(), 'project', None)) or {}
                if isinstance(overlay_map, dict) and overlay_map:
                    # Only add overlay_map keys if they're explicitly requested in fields_for_evaluation
                    # or if all_fields=True. Don't automatically add all overlay_map keys to avoid
                    # processing all tasks when only a page is needed (e.g., in only_filtered).
                    # Merge overlay with base map for this call only (all keys available, but only used if requested)
                    annotations_map = {**annotations_map, **overlay_map}
                    # Only add overlay_map keys to fields_for_evaluation if they're explicitly requested
                    if fields_for_evaluation is not None:
                        # Only include overlay_map keys that are already in fields_for_evaluation
                        overlay_keys_in_request = [k for k in overlay_map.keys() if k in fields_for_evaluation]
                        if overlay_keys_in_request:
                            # Ensure they're in the list (they already are, but this makes it explicit)
                            fields_for_evaluation = list(set(fields_for_evaluation) | set(overlay_keys_in_request))

        if fields_for_evaluation is None:
            fields_for_evaluation = []

        if excluded_fields_for_evaluation is None:
            excluded_fields_for_evaluation = []

        first_task = queryset.first()
        project = None if first_task is None else first_task.project

        # db annotations applied only if we need them in ordering or filters
        for field in annotations_map.keys():
            # Include field if it's explicitly requested or all_fields=True, but exclude if it's in the exclusion list
            if (field in fields_for_evaluation or all_fields) and field not in excluded_fields_for_evaluation:
                queryset.project = project
                queryset.request = request
                function = annotations_map[field]
                queryset = function(queryset)