Workflow Query Architecture

Developer guide for the Enezaframework workflow engine. Part of the Workflow series.

Overview
The Two Enhancements
Component Map
Query Execution Pipeline
Object ID Data Flow
Aggregate Data Flow
Related Guides

Overview

The workflow query enhancements extend SavedQuery (the stored query-builder condition) in two ways:

Object ID filtering — get_queryset(object_id=...) scopes results to one record, so condition steps match only the document being evaluated.
Aggregate annotations — an annotations array in filters_json adds SUM/COUNT/AVG/MIN/MAX over related models and lets filter rules compare those computed values.

Both are additive: callers that pass no object_id and store no annotations behave exactly as before.

The Two Enhancements

Enhancement	Entry point	File
Object ID filtering	`SavedQuery.get_queryset(object_id=None)`	`enezaworkflowengine/models/querybuilder.py:31`
Aggregate annotations	`parse_annotations(model_class, annotations_list)`	`enezaworkflowengine/utils.py:189`

Component Map

flowchart TD
    subgraph Model["SavedQuery"]
        GQ["get_queryset(object_id=None)<br/>models/querybuilder.py:31"]
        EX["execute(object_id=None)<br/>:62"]
        ENF["execute_new_format(object_id=None)<br/>:69"]
        FJ["filters_json<br/>(rules / groups / annotations)"]
    end

    subgraph Utils["enezaworkflowengine/utils.py"]
        PA["parse_annotations()<br/>:189"]
        MAP["AGGREGATE_FUNCTION_MAP<br/>:180"]
        VAL["_validate_relationship()<br/>:245"]
    end

    subgraph Engine["Workflow engine"]
        CS["evaluate_condition_for_document()<br/>services/condition_step.py:16"]
        SH["ConditionStepHandler._evaluate_and_route()<br/>services/step_handlers.py:233"]
        TASK["evaluate_condition_step()<br/>tasks/workflow_tasks.py:216"]
    end

    subgraph UI["Query builder UI"]
        MIX["QueryBuilderMixin.get_aggregate_options()<br/>views/mixin.py:152"]
        BQV["BuildQueryView<br/>views/querybuilder.py:55"]
        SQD["SaveQueryDetailView<br/>views/querybuilder.py:180"]
    end

    FJ --> GQ
    MAP --> PA
    VAL --> PA
    PA --> GQ
    GQ --> CS
    GQ --> SH
    CS --> TASK
    MIX --> BQV
    MIX --> SQD
    BQV --> FJ

Query Execution Pipeline

SavedQuery.get_queryset() applies annotations before filters, then applies the optional object ID (models/querybuilder.py:31-60).

flowchart TD
    START["get_queryset(object_id=None)"] --> MODEL["model_class = content_type.model_class()"]
    MODEL --> HASANN{"filters_json has 'annotations'?"}
    HASANN -->|Yes| ANN["parse_annotations() then annotate(**ann)"]
    HASANN -->|No| ALL["model_class.objects.all()"]
    ANN --> PARSER{"'groups' in filters_json?"}
    ALL --> PARSER
    PARSER -->|Yes| NEW["parse_new_query_format_to_q()"]
    PARSER -->|No| OLD["parse_rules_to_q()"]
    NEW --> FILTER["qs.filter(q)"]
    OLD --> FILTER
    FILTER --> OBJ{"object_id is not None?"}
    OBJ -->|Yes| PK["qs.filter(pk=object_id)"]
    OBJ -->|No| RETURN["Return queryset"]
    PK --> RETURN

execute() and execute_new_format() forward object_id to the same two-stage filtering. execute_new_format() is the variant used when the stored query may use the groups format (models/querybuilder.py:69-103).

Object ID Data Flow

A WorkflowInstance already holds content_type and object_id, and exposes the target record as document (a GenericForeignKey). Condition evaluation therefore never needs the caller to supply an ID.

sequenceDiagram
    participant WI as WorkflowInstance
    participant SH as ConditionStepHandler
    participant CS as evaluate_condition_for_document
    participant T as Celery task
    participant SQ as SavedQuery

    WI->>SH: _evaluate_and_route(step_instance, workflow_instance, ...)
    SH->>SQ: get_queryset(object_id=workflow_instance.object_id)
    SQ-->>SH: scoped queryset
    SH-->>WI: CONDITION_TRUE / CONDITION_FALSE

    WI->>T: evaluate_condition_step(step_instance_id)
    T->>CS: evaluate_condition_for_document(workflow_instance.document, query)
    CS->>CS: object_id = str(document.pk)
    CS->>SQ: get_queryset(object_id=object_id)
    SQ-->>CS: scoped queryset
    CS-->>T: bool(count > 0)

See Workflow Object ID Filtering and Workflow Engine Auto-Extraction.

Aggregate Data Flow

An annotation entry becomes a Django aggregate, and its alias becomes available to filter rules.

flowchart LR
    JSON["filters_json.annotations[]"] --> PA["parse_annotations()"]
    PA --> VALID["_validate_relationship()"]
    VALID --> EXPR["{alias: Sum('lines__amount'), ...}"]
    EXPR --> ANNOTATE["model.objects.annotate(**expr)"]
    RULES["filters_json.rules[]<br/>field = alias"] --> Q["parse_rules_to_q()"]
    ANNOTATE --> QFILTER["queryset.filter(q)"]
    Q --> QFILTER
    QFILTER --> HAVING["Django pushes aggregate<br/>filters to HAVING"]

Example stored query:

json

{
  "condition": "AND",
  "annotations": [
    {"function": "sum", "field": "amount", "related_model": "lines", "alias": "total_amount"}
  ],
  "rules": [
    {"id": "total_amount", "field": "total_amount", "type": "double", "operator": "greater", "value": "1000"}
  ]
}

parse_annotations() validates that related_model is a real relationship, builds the lookup lines__amount, and returns {'total_amount': Sum('lines__amount')} (utils.py:189-242). See Workflow Aggregate Annotations.