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.

📘

Prerequisites

Build 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
FieldValuesNotes
output_detailOutputDetail.DECISIONS, OutputDetail.BITMASKSUse DECISIONS for routing; BITMASKS only when you need per-rule masks.
ingest_error_modeIngestErrorMode.SKIP_AND_COUNT (default), SKIP_TO_DEAD_LETTER, HARD_FAILHow malformed records are handled.
type_mismatch_modeTypeMismatchMode.NULL_ON_TYPE_ERROR (default), COERCE, HARD_FAIL_TYPEHow a wrong-typed field value is handled.
simd_backend_override"auto", "scalar", "neon", "sse2", "avx2", "avx512"Force a backend; "auto" selects at runtime.
enable_avx512True / FalseAVX-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 inference

You 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.

MethodPurpose
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()
FieldMeaning
n_recordsRecords evaluated in the batch.
n_matchedRecords that matched at least one rule.
decisionsPer-record decision label (e.g. APPROVE, REVIEW, BLOCK).
decision_codesPer-record integer code for the decision.
scoresPer-record accumulated score.
risk_bandsPer-record risk band.
winning_rule_idsThe rule that won precedence per record.
match_countsPer-rule match count across the batch.
matched_indicesIndices of records that matched any rule.
timing_msStage/total timing for the batch.
messages_processed / messages_skippedIngest counters.
error_counts / error_samplesIngest 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 / fieldPurpose
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_model and model_score need a build with BLAZERULES_ENABLE_ONNX=ON (the default). A lean build without ONNX rejects model_score rules 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_ERROR

Python 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")

Where to go next