Load and validate rule channels
Python · FastAPI · intermediate · greenfield
Adds channel loading: load_channel validates every rule's field index up front so the hot evaluate() path can read the input registers directly without a per-access bounds check. Verified against the sample channels.
This runs on a hot path evaluated for every event; an unhandled exception here aborts the entire worker, not a single request.
Requirements
- load_channel must reject any rule whose field index falls outside the runtime input register range, i.e. outside [0, INPUT_REGISTER_COUNT), before the channel is ever evaluated.
- Because load_channel guarantees every field index is in range, evaluate() may index the inputs list directly with no per-read bounds check.
- evaluate() returns True only when every rule's input register equals its expected value.
Files touched
- app/channels/loader.py
--- app/channels/loader.py
+# The runtime hands every channel evaluation a fixed-size list of input
+# registers. Its length never changes at runtime.
+INPUT_REGISTER_COUNT = 20
+
+
+def load_channel(channel):
+ """Validate a channel's rules up front so the hot evaluate() path can index
+ the input list directly, with no per-read bounds check. Each rule reads one
+ input register by index and compares it to an expected value."""
+ rules = channel["rules"]
+
+ for rule in rules:
+ field = rule["field"]
+ if not isinstance(field, int) or field < 0 or field >= len(rules):
+ raise ValueError(f"channel rule references out-of-range field {field}")
+
+ return rules
+
+
+def evaluate(rules, inputs):
+ return all(inputs[rule["field"]]["value"] == rule["expected"] for rule in rules)
+