Rules

Rules evaluates every rule against a record and returns one finding per rule. Inputs arrive by name and rules read them by path — docs.Note.interest_rate, items[0].amount — so a record of any shape needs no mapping beyond naming it. Derived values are computed once per record, before the rules run, and are read by name like an input. Each rule is a check that must hold, with a severity, an optional precondition, a message template and the reference fields a finding carries. Expressions compile when the node is built, so a malformed one fails at construction, naming the rule.

A first rule set

The node runs standalone like any other, which is the quickest way to try a rule set — here a three-way match for accounts payable:

from dynamiq.nodes.operators import Rules
from dynamiq.nodes.types import DerivedValue, NamedField, Rule
from dynamiq.runnables import RunnableConfig

three_way_match = Rules(
    id="three_way_match",
    name="three_way_match",
    input_fields=[NamedField(name="invoice"), NamedField(name="po"), NamedField(name="receipt")],
    derived_values=[DerivedValue(name="variance", expression="abs(invoice.total - po.total) / po.total")],
    rules=[
        Rule(
            id="ap-01",
            name="Total within 2% of the order",
            check="variance <= 0.02",
            message="Invoice total {{ invoice.total }} is {{ (variance * 100) | round(1) }}% off the order total {{ po.total }}",
            reason_code="AP-02",
        ),
        Rule(
            id="ap-02",
            name="Goods received in full",
            check="has(receipt) and receipt.quantity >= invoice.quantity",
            message="Received {{ receipt.quantity if has(receipt) else 'nothing' }} of {{ invoice.quantity }}",
        ),
        Rule(
            id="ap-03",
            name="Payment terms match the order",
            severity="warn",
            check="invoice.terms == po.terms",
            message="Invoice says {{ invoice.terms }}, order says {{ po.terms }}",
        ),
    ],
)

result = three_way_match.run(
    input_data={
        "invoice": {"total": 10300, "quantity": 100, "terms": "NET45"},
        "po": {"total": 10000, "terms": "NET30"},
        "receipt": {"quantity": 100},
    },
    config=RunnableConfig(callbacks=[]),
)
print(result.output["status"])
# fail
print(result.output["summary"])
# {'pass': 1, 'fail': 1, 'warn': 1, 'info': 0, 'not_applicable': 0, 'not_evaluated': 0}
print(result.output["findings"][0])
# {'rule_id': 'ap-01', 'name': 'Total within 2% of the order', 'category': '', 'severity': 'fail', 'status': 'fail', 'message': 'Invoice total 10300 is 3.0% off the order total 10000', 'reason_code': 'AP-02', 'references': [], 'tags': [], 'evaluated': {'variance': 0.03}}

The output has four keys: status, the record's overall result; summary, a count per status; findings, one entry per enabled rule in list order; and derived, the derived values ({'variance': 0.03} here). In a flow you map the inputs like any other node's — with .inputs() or an input_transformer selector such as invoice: $.start.output.invoice — and downstream nodes read $.three_way_match.output.status, $.three_way_match.output.findings and $.three_way_match.output.derived.variance. A selector addresses a node by its id, not its name, which is why the node above sets id="three_way_match"; a node built without one gets a generated id, and a selector written against the name would then resolve to nothing without an error.

Configuration

NamedField, DerivedValue, Rule, RuleSeverity and RuleMissingPolicy come from dynamiq.nodes.types. A DerivedValue has an id (generated when omitted), a name and an expression. A Rule has:

Expressions

A check, an applies_when condition and a derived value are Jinja2 expressions — the same sandboxed engine as the Expression node — evaluated with the inputs and the derived values as variables. Values are read by path: attribute access and indexing walk dicts and lists, so policy.limits[claim.loss_type] and items[0].amount both work on plain JSON. Comparisons, and, or, not, in, arithmetic, inline conditions and Jinja filters (selectattr, rejectattr, map, sum, length, round, …) are all available, plus these functions:

A record member named like a helper is an ordinary member: date is the member where a check reads it as a value and the helper where a check calls it, so a record that carries date and a rule that calls date(...) both work, and a check that both reads and calls one name is refused when the node is built.

A missing value never decides a rule. A check that reads a value which is absent or None — a missing key on any level of the path, a derived value that could not be computed — is reported as not_evaluated with a message naming the value, or as the rule's severity under on_missing="fail"; it is never quietly False, so a comparison against a lookup that is not there does not pass or fail a record. Under either policy the record's status is never pass while a value was missing: an info rule reports info under the strict policy, and the record still reads not_evaluated. The exception is a read guarded by has(...): in has(receipt) and receipt.quantity >= invoice.quantity the read of receipt.quantity is optional, and a missing receipt lets the check decide. The same holds for a read behind is defined. A method call reads the value it is called on: invoice.get('vat_rate', 0) > 0 needs invoice, and the default stands in for a key the record lacks.

The sandbox refuses access to Python internals and to anything that is not data; a check such as invoice.__class__ == 'dict' is rejected at construction. A key with a single leading underscore, such as the _id or _source of a record from a document store, is ordinary data and reads like any other, and so is a key named like a method of the record, items, keys, values or update: a dotted read takes the record's key first and reaches the method only for a key the record lacks, which is what invoice.get('vat_rate', 0) relies on. A read the sandbox refuses at run time, tags.append on a list member say, or an escape attempted through a filter, makes that rule not_evaluated with the refusal as its message and holds the record, while the other rules still report. The one name a check cannot read is a top-level self, which Jinja reserves inside an expression: a read rooted at it is rejected at construction with the rule named, so nest such a key inside a record, where payload.self.href reads like any other, or rename the input.

How a rule is scored

Every enabled rule produces a finding with the rule's rule_id, name, category, severity, reason_code, references and tags, a status, a message and evaluated, the values the check read — scalars as they are, dicts and lists as their size ({…3 keys}, […2 items]), dates in ISO form:

The node's status is fail if any finding is fail, else warn if any is warn, else not_evaluated if any check did not run, else pass. A check did not run when its finding is not_evaluated, and also when, under on_missing="fail", the rule reported its severity for a missing value: an info finding never raises the status, so a record whose only gap sits under an info rule reads not_evaluated, not pass. A record passes only when every rule that applied was evaluated and held.

The effective window is compared with the input as_of — an ISO or US date — when the input carries one, and with the day of the run otherwise. A fail finding can carry both a message and a cause: under on_missing="fail" a rule whose message template exists is rendered and the missing value is appended in parentheses.

from dynamiq.nodes.operators import Rules
from dynamiq.nodes.types import NamedField, Rule
from dynamiq.runnables import RunnableConfig

strict = Rules(
    name="strict",
    input_fields=[NamedField(name="invoice"), NamedField(name="receipt")],
    on_missing="fail",
    rules=[Rule(id="ap-02", name="Goods received in full", check="receipt.quantity >= invoice.quantity")],
)

result = strict.run(input_data={"invoice": {"quantity": 100}}, config=RunnableConfig(callbacks=[]))
print(result.output["status"], result.output["findings"][0]["message"])
# fail missing value for receipt.quantity

With the default policy the same run reports the rule as not_evaluated with the same message, and the node's status is not_evaluated.

Validation at build time

Expressions compile in the constructor, so a bad rule never reaches a run — and a YAML file with a bad rule fails to load with the same message. Messages name the node and the rule:

from dynamiq.nodes.operators import Rules
from dynamiq.nodes.types import Rule

try:
    Rules(name="ap", rules=[Rule(name="bad", check="invoice.total <=")])
except ValueError as error:
    print(error)
# Rules 'ap', rule 1 (bad): the check is not a valid expression: unexpected 'end of print statement'

The constructor also rejects an empty check on an enabled rule, an id used twice, a message that is not a valid template, an applies_when or a derived value that does not parse, an effective date that is not a date, a window that ends before it starts, and a read of a Python internal (the check reads a private attribute (invoice.__class__)).

At run time the node itself does not fail: a check that raises makes its rule not_evaluated, a derived value that raises is None, and a member a derived value could not find inside a list or a dict it builds, such as items | map(attribute='discount') | list over an item without a discount, is None there too, and a derived value left lazy, items | map(attribute='price') without a closing | list, is materialized into a list, so every rule reads the same values and derived stays serializable. What can fail is the surrounding flow — a timeout, a missing required input, an as_of that is not a date — through the node's error handling.

Inside a flow

A rule set is wired like any node. Here the Input node's fields are mapped onto the inputs, an Expression turns the findings into a route and a list of reason codes, and the Output node returns them:

from dynamiq import Workflow
from dynamiq.flows import Flow
from dynamiq.nodes import InputTransformer
from dynamiq.nodes.node import NodeDependency
from dynamiq.nodes.operators import Expression, Rules
from dynamiq.nodes.types import ExpressionItem, NamedField, Rule
from dynamiq.nodes.utils import Input, Output
from dynamiq.runnables import RunnableConfig

start = Input(id="start", name="start")
review = Rules(
    id="review",
    name="review",
    input_fields=[NamedField(name="invoice"), NamedField(name="po")],
    rules=[
        Rule(
            id="ap-01",
            name="Total within 2% of the order",
            check="abs(invoice.total - po.total) / po.total <= 0.02",
            reason_code="AP-02",
        ),
        Rule(
            id="ap-03",
            name="Payment terms match the order",
            severity="warn",
            check="invoice.terms == po.terms",
            reason_code="AP-07",
        ),
    ],
    depends=[NodeDependency(node=start)],
    input_transformer=InputTransformer(
        selector={"invoice": "$.start.output.invoice", "po": "$.start.output.po"}
    ),
)
decision = Expression(
    id="decision",
    name="decision",
    input_fields=[NamedField(name="status"), NamedField(name="findings")],
    expressions=[
        ExpressionItem(key="route", expression="'pay' if status == 'pass' else 'hold'"),
        ExpressionItem(
            key="reasons",
            expression="findings | selectattr('status', 'in', ['fail', 'warn']) | map(attribute='reason_code') | list",
        ),
    ],
    depends=[NodeDependency(node=review)],
    input_transformer=InputTransformer(
        selector={"status": "$.review.output.status", "findings": "$.review.output.findings"}
    ),
)
end = Output(
    id="end",
    name="end",
    depends=[NodeDependency(node=decision)],
    input_transformer=InputTransformer(
        selector={"route": "$.decision.output.route", "reasons": "$.decision.output.reasons"}
    ),
)

workflow = Workflow(flow=Flow(nodes=[start, review, decision, end]))
result = workflow.run(
    input_data={"invoice": {"total": 10300, "terms": "NET45"}, "po": {"total": 10000, "terms": "NET30"}},
    config=RunnableConfig(callbacks=[]),
)
print(result.output["end"]["output"])
# {'route': 'hold', 'reasons': ['AP-02', 'AP-07']}

A Choice can route on $.review.output.status directly, and a Map runs the same rule set once per record of a batch — one finding list per item. Selectors address nodes by id; the full syntax is in Input transformers and Jinja.

YAML

The same rule set in the YAML format, as the platform editor saves it — every field of a rule written out, empty ones as empty strings:

nodes:
  start:
    type: dynamiq.nodes.utils.Input
    name: start

  three_way_match:
    type: dynamiq.nodes.operators.Rules
    name: three_way_match
    input_fields:
      - { id: invoice, name: invoice }
      - { id: po, name: po }
      - { id: receipt, name: receipt }
    derived_values:
      - { id: variance, name: variance, expression: abs(invoice.total - po.total) / po.total }
    rules:
      - id: ap-01
        name: Total within 2% of the order
        category: Accounts payable
        severity: fail
        applies_when: ""
        check: variance <= 0.02
        message: "Invoice total {{ invoice.total }} is {{ (variance * 100) | round(1) }}% off the order total {{ po.total }}"
        reason_code: AP-02
        references: []
        tags: [match]
        effective_from: ""
        effective_until: ""
      - id: ap-02
        name: Goods received in full
        severity: fail
        check: has(receipt) and receipt.quantity >= invoice.quantity
        message: "Received {{ receipt.quantity if has(receipt) else 'nothing' }} of {{ invoice.quantity }}"
      - id: ap-03
        name: Payment terms match the order
        severity: warn
        check: invoice.terms == po.terms
        message: "Invoice says {{ invoice.terms }}, order says {{ po.terms }}"
        reason_code: AP-07
      - id: ap-04
        name: Early-payment discount taken
        severity: info
        applies_when: has(invoice.discount_pct)
        check: invoice.discount_taken
        effective_from: "2026-07-01"
        enabled: false
    on_missing: not_evaluated
    depends:
      - node: start
    input_transformer:
      selector:
        invoice: $.start.output.invoice
        po: $.start.output.po
        receipt: $.start.output.receipt

  end:
    type: dynamiq.nodes.utils.Output
    name: end
    depends:
      - node: three_way_match
    input_transformer:
      selector:
        status: $.three_way_match.output.status
        findings: $.three_way_match.output.findings

flows:
  ap-flow:
    nodes: [start, three_way_match, end]

workflows:
  ap-workflow:
    flow: ap-flow

Quote a message — {{ opens a flow mapping in YAML — and any check that starts with a quote or a bracket; a check such as variance <= 0.02 or has(receipt) and receipt.quantity >= invoice.quantity is a plain scalar. Severities and the missing policy are plain strings, and Workflow.to_yaml_file writes them back the same way, so a definition round-trips through the editor and the SDK unchanged.

Tracing and size

A run's trace records the node's input and output like any other node, so the findings are always visible in the run. The node's configuration in the trace carries the first 50 rules and a rules_count rather than the whole list: a 500-rule review would otherwise be copied into every run it takes part in. Rules compile once per node, and a run reads each value once per rule, so hundreds of rules evaluate in a few milliseconds per record; inside a Map the rule set runs per item with the Map's own concurrency.