]j|dZddlZddlmZddlmZddlmZmZddl m Z eje Z GddZy) a FSM QuerySet Mixins for annotating entities with their current state. Provides reusable Django QuerySet mixins that efficiently annotate entities with their current FSM state using optimized subqueries to prevent N+1 queries. Usage: class TaskQuerySet(FSMStateQuerySetMixin, models.QuerySet): pass class TaskManager(models.Manager): def get_queryset(self): return TaskQuerySet(self.model, using=self._db) Note: State annotation is guarded by 'fflag_feat_fit_568_finite_state_management' only. This allows background FSM processes to annotate current_state for internal use. UI/API consumption of state fields is separately controlled by serializers that check BOTH 'fflag_feat_fit_568_finite_state_management' AND 'fflag_feat_fit_710_fsm_state_fields' before exposing state data. When the flag is disabled, no annotation is performed and there is zero performance impact. N)CurrentContext)flag_set)OuterRefSubquery)get_state_modelceZdZdZdZy)FSMStateQuerySetMixina. Mixin for Django QuerySets to efficiently annotate FSM state. Provides the `with_state()` method that adds a `current_state` annotation to the queryset using an optimized subquery. This approach: - Prevents N+1 queries by using a single JOIN/subquery - Handles missing states gracefully (returns None) - Uses UUID7 natural ordering for optimal performance - Works with any FSM entity that has a registered state model Example: # In your model manager class TaskManager(models.Manager): def get_queryset(self): return TaskQuerySet(self.model, using=self._db) def with_state(self): return self.get_queryset().with_state() # Usage - both approaches work identically tasks = Task.objects.with_state().filter(project=project) # Or chain it after filters tasks = Task.objects.filter(project=project).with_state() for task in tasks: print(f"Task {task.id}: {task.current_state}") # No additional queries! c tj}td|stj d|S|j j j}t|}|stj d|d|S|j}|d}t|jjd i|tdijdjd d d }|j!| S)a Annotate the queryset with the current FSM state. Adds a `current_state` field to each object containing the current state string value. This is done using an efficient subquery that leverages UUID7 natural ordering to prevent N+1 queries. Returns: QuerySet: The annotated queryset with `current_state` field Example: # Chain after filters tasks = Task.objects.filter(project=project).with_state() # Or use from manager tasks = Task.objects.with_state().filter(project=project) # Multiple chaining tasks = Task.objects.filter(is_labeled=True).with_state().order_by('-created_at') Note: - If FSM feature flag is disabled, returns queryset unchanged (zero impact) - If no state exists for an entity, `current_state` will be None - The state is read-only and should not be modified directly *fflag_feat_fit_568_finite_state_management)userz4FSM feature flag disabled, skipping state annotationzNo state model registered for z, skipping annotation_idpkz-idstateN) current_state)rget_userrloggerdebugmodel_meta model_namer_get_entity_field_namerobjectsfilterrorder_byvaluesannotate)selfr entity_name state_modelentity_field_namefk_fieldcurrent_state_subquerys J/root/env/lib/python3.12/site-packages/label_studio/fsm/queryset_mixins.py with_statez FSMStateQuerySetMixin.with_stateCs<&&(D4P LLO PKjj&&11 &k2  LL9+F[\ ]K(>>@'(,"* &K   & & D(HTN)C D M Me T [ [\c degfg h"  }}+A}BBN)__name__ __module__ __qualname____doc__r&rr'r%r r $s <=Cr'r )r+loggingcore.current_requestrcore.feature_flagsrdjango.db.modelsrr fsm.registryr getLoggerr(rr rr'r%r2s;2/'/(   8 $\C\Cr'