Python API
The blazerules Python module: configure an engine, load rules, evaluate batches, and read decisions, scores, and routing groups.
The blazerules Python module wraps the C++ core via pybind11. The API is batch-first: you accumulate records, then evaluate a whole batch at once. There is no per-record Python evaluation path.
PrerequisitesBuild the Python module and put it on
PYTHONPATH(see Installation). Verify with:python -c "import blazerules; print(blazerules.__version__, blazerules.simd_backend())"
Engine configuration
blazerules.EngineConfig() holds engine-wide settings. The most common is output_detail.
import blazerules
config = blazerules.EngineConfig()
config.output_detail = blazerules.OutputDetail.DECISIONS| Field | Values | Notes |
|---|---|---|
output_detail | OutputDetail.DECISIONS, OutputDetail.BITMASKS | Use DECISIONS for routing; BITMASKS only when you need per-rule masks. |
ingest_error_mode | IngestErrorMode.SKIP_AND_COUNT (default), SKIP_TO_DEAD_LETTER, HARD_FAIL | How malformed records are handled. |
type_mismatch_mode | TypeMismatchMode.NULL_ON_TYPE_ERROR (default), COERCE, HARD_FAIL_TYPE | How a wrong-typed field value is handled. |
simd_backend_override | "auto", "scalar", "neon", "sse2", "avx2", "avx512" | Force a backend; "auto" selects at runtime. |
enable_avx512 | True / False | AVX-512 is off for auto-selection unless enabled (see Configuration Reference). |
Constructing an engine
There are three ways to construct a RuleEngine:
# 1. Defaults
engine = blazerules.RuleEngine()
# 2. With a config
engine = blazerules.RuleEngine(config)
# 3. With an explicit schema (skip inference)
schema = [
blazerules.Field("card_token", ...),
# ...
]
engine = blazerules.RuleEngine(schema, config)
Schema inferenceYou can
load_rules(...)before any schema exists. The first evaluated batch samples rule-referenced fields, infers supported types, binds the schema, then compiles and activates the rules. Pass an explicit schema only when you want full control. See Data Model & Schema.
Loading rules
engine.load_rules("rules.yaml")Rules and schema activation are strict: bad YAML, unknown fields, duplicate rule IDs, invalid regex, bad lookup files, and type/operator mismatches fail before activation. See Error Reference.
Rules can also be loaded from an exact-object s3:// URI (see S3 resources).
RuleEngine reference
These are the public methods exposed by the Python binding. The hot path is still batch-shaped: every evaluation method accepts a batch, file, or contiguous byte buffer.
| Method | Purpose |
|---|---|
load_rules(path) | Load YAML/JSON rules from local path or exact-object s3:// URI. |
load_rules_from_string(text) | Load rules from an in-memory YAML/JSON string. |
reload_rules_now(path) | Compile and atomically swap a new ruleset immediately. |
analyze_conflicts(path) | Parse and analyze a ruleset without activating it. |
enable_hot_reload(path, poll_interval_seconds=5) | Watch a rules file and reload only after successful validation. |
stop_hot_reload() | Stop the watcher thread. |
hot_reload_status() | Return active version, attempts, successes, failures, and last error. |
evaluate_ndjson(bytes) | Evaluate newline-delimited JSON bytes. |
evaluate_ndjson_padded(payload, logical_size) | Evaluate already-padded NDJSON without an extra copy. |
evaluate_ndjson_file(path) | Memory-map and evaluate an NDJSON file. |
evaluate_messages(list[str]) | Compatibility path for lists of JSON messages; prefer evaluate_ndjson for high throughput. |
evaluate_batch(pyarrow.RecordBatch) | Evaluate a typed Arrow batch, including projected nested structs. |
create_shards(shard_count) | Create partition-affine engines for streaming/window workloads. |
evaluate_partition_messages(partition_id, messages) | Evaluate a message batch on a partition shard. |
evaluate_partition_ndjson_padded(partition_id, payload, logical_size) | Evaluate padded NDJSON on a partition shard. |
evaluate_partition_batch(partition_id, batch) | Evaluate an Arrow batch on a partition shard. |
backtest(parquet_path, rules_a, rules_b, label_column=None) | Compare two rulesets over one or more Parquet files. |
reset_window_state() | Clear in-memory window state. |
num_window_channels() | Return the number of compiled window channels. |
register_model(name, path) | Register an ONNX model for model_score rules. |
num_models() | Return registered model count. |
enable_metrics() / reset_metrics() / metrics_enabled() / metrics_snapshot() | In-process metrics collection. |
active_rule_set_version() | Return the active ruleset version string. |
Evaluating a batch
Pick the entry point that matches your input. All return a result object.
payload = b"""
{"card_token":"card_1","amount":2500.0,"device_type":"emulator",
"country_code":"US","account_age_days":2,"hour_of_day":1.5}
{"card_token":"card_2","amount":50.0,"device_type":"ios",
"country_code":"GB","account_age_days":400,"hour_of_day":12}
"""
result = engine.evaluate_ndjson(payload)
print(result.n_records, result.n_matched)
print(result.decisions)
print(result.match_counts)Reading results
The result object exposes batch-shaped outputs. Prefer the decision-group accessors over Python loops.
result = engine.evaluate_ndjson(payload)
approved = result.indices_for_decision("APPROVE")
needs_review = result.indices_for_not_decision("APPROVE")
groups = result.grouped_decision_indices()| Field | Meaning |
|---|---|
n_records | Records evaluated in the batch. |
n_matched | Records that matched at least one rule. |
decisions | Per-record decision label (e.g. APPROVE, REVIEW, BLOCK). |
decision_codes | Per-record integer code for the decision. |
scores | Per-record accumulated score. |
risk_bands | Per-record risk band. |
winning_rule_ids | The rule that won precedence per record. |
match_counts | Per-rule match count across the batch. |
matched_indices | Indices of records that matched any rule. |
timing_ms | Stage/total timing for the batch. |
messages_processed / messages_skipped | Ingest counters. |
error_counts / error_samples | Ingest error tallies and samples. |
See Decisions & Scoring for how these are produced and how to route on them.
Result routing helpers
Use these helpers to avoid scanning large Python string arrays row by row:
| Method / field | Purpose |
|---|---|
indices_for_decision("APPROVE") | Row indices with a specific final decision. |
indices_for_not_decision("APPROVE") | Row indices with any other decision. |
grouped_decision_indices() | Mapping from decision label to row-index array. |
indices_for_rule("rule_id") | Row indices where a rule fired. |
grouped_winning_rule_indices() | Mapping from winning rule id to row-index array. |
result["rule_id"] | Per-rule NumPy boolean mask when OutputDetail.BITMASKS is enabled. |
Hot reload
engine.load_rules("rules.yaml")
engine.enable_hot_reload("rules.yaml", poll_interval_seconds=5)
status = engine.hot_reload_status()New YAML/lookups are compiled and validated off the hot path, then swapped atomically only on success; failed reloads keep the previous ruleset. See Hot Reload.
ML model registration
engine.register_model("fraud_logreg", "models/fraud_logreg.onnx")Then reference the model from a model_score rule (see ML Scoring).
Requires ONNX
register_modelandmodel_scoreneed a build withBLAZERULES_ENABLE_ONNX=ON(the default). A lean build without ONNX rejectsmodel_scorerules at compile time.
Diagnostics & compatibility
import blazerules
print(blazerules.simd_backend()) # active backend, e.g. "neon"
print(blazerules.cpu_features_summary()) # detected CPU features
print(blazerules.__version__) # library version
print(blazerules.RULE_YAML_COMPATIBILITY) # supported YAML major, e.g. "2.x"Error handling
Ingest defaults are tolerant; activation is strict. Configure per workload:
config.ingest_error_mode = blazerules.IngestErrorMode.SKIP_AND_COUNT
config.type_mismatch_mode = blazerules.TypeMismatchMode.NULL_ON_TYPE_ERRORPython raises typed errors on activation failures: BlazeRulesError, BlazeRulesConfigError, BlazeRulesParseError, BlazeRulesSchemaError. See Error Reference.
S3 resources
Rules, lookups, and models can be loaded from exact-object s3://bucket/key URIs via the AWS CLI cache path.
import blazerules
blazerules.set_aws_profile("personal")
blazerules.set_aws_region("us-east-1")
blazerules.set_aws_endpoint_url("http://127.0.0.1:9000") # MinIO / LocalStack / R2
engine = blazerules.RuleEngine()
engine.load_rules("s3://bucket/rules/fraud.yaml")