E

Workflow Engine Auto-Extraction

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

Table of Contents

  1. Overview
  2. WorkflowInstance Holds the Target
  3. Web Path — _evaluate_and_route()
  4. Celery Path — evaluate_condition_step()
  5. Service Function — evaluate_condition_for_document()
  6. Related Guides

Overview

Callers never pass object_id to condition queries. A WorkflowInstance already stores the target document, so the engine extracts the primary key and passes it to SavedQuery.get_queryset(). The extraction happens in two code paths, both ending in the same scoped filtering.

flowchart TD
    WI["WorkflowInstance<br/>content_type + object_id + document"] --> W["Web path:<br/>ConditionStepHandler._evaluate_and_route()"]
    WI --> C["Celery path:<br/>evaluate_condition_step()"]
    W --> WQ["get_queryset(object_id=wf_instance.object_id)"]
    C --> S["evaluate_condition_for_document(document, query)"]
    S --> SQ["object_id = str(document.pk)"]
    SQ --> CQ["get_queryset(object_id=object_id)"]

WorkflowInstance Holds the Target

Defined in enezaworkflowengine/models/workflowtemplate.py:104, WorkflowInstance stores:

python
object_id = models.TextField(null=True, blank=True)
document = GenericForeignKey('content_type', 'object_id')

(models/workflowtemplate.py:124-125). So workflow_instance.object_id and workflow_instance.document.pk are two routes to the same primary key.

Web Path — _evaluate_and_route()

File: enezaworkflowengine/services/step_handlers.py:233

The condition step handler has direct access to workflow_instance.object_id and passes it without an extra query:

python
# services/step_handlers.py:233-243
def _evaluate_and_route(self, step_instance, workflow_instance, step_template):
    condition_query = step_template.condition_expression
    # ...
    queryset = condition_query.get_queryset(object_id=workflow_instance.object_id)
    result = queryset.count() > 0

Celery Path — evaluate_condition_step()

File: enezaworkflowengine/tasks/workflow_tasks.py:216

The background task delegates to the service function, which extracts the PK from the resolved document:

python
# tasks/workflow_tasks.py:253-254
from enezaworkflowengine.services.condition_step import evaluate_condition_for_document
result = evaluate_condition_for_document(workflow_instance.document, condition_query)

The task itself needed no change: the extraction lives in the service function, so both paths share it.

Service Function — evaluate_condition_for_document()

File: enezaworkflowengine/services/condition_step.py:16

python
def evaluate_condition_for_document(document, condition_query):
    if not document or not condition_query:
        return False

    try:
        object_id = str(document.pk)
        queryset = condition_query.get_queryset(object_id=object_id)
        result_count = queryset.count()
        return result_count > 0
    except Exception as e:
        raise ConditionStepError(f"Failed to evaluate condition: {str(e)}")

This function is also usable outside the engine — for previewing or testing a condition against a specific document.

sequenceDiagram
    participant T as evaluate_condition_step (Celery)
    participant S as evaluate_condition_for_document
    participant Q as SavedQuery

    T->>S: (workflow_instance.document, condition_query)
    S->>S: object_id = str(document.pk)
    S->>Q: get_queryset(object_id=object_id)
    Q-->>S: queryset
    S->>S: return count > 0
    S-->>T: bool
View source Edit this page