rudy-ab-test

A lightweight, rule-based feature-flag and A/B test engine for Python.
Define variables with fallback defaults, attach prioritized rules driven by arbitrary context dicts, and evaluate them at runtime — no external services required.

Installation

pip install rudy-ab-test

Quick start

from rudy_ab_test import RuleEngine, Variable, Rule, Condition

engine = RuleEngine()

rule = Rule(
    name="premium_model_for_prod",
    conditions=[Condition(field="environment", operator="eq", value="prod")],
    result="gpt-4",
    priority=1,
)

engine.register(Variable(name="model_name", default="gpt-3.5-turbo", rules=[rule]))

result = engine.evaluate("model_name", {"environment": "prod"})
# → "gpt-4"

Core concepts

Condition

A single predicate evaluated against the context dict.

Supported operators

Rule

A rule fires when all its conditions match. The first matching rule (by descending priority) determines the variable's value.

Variable

A named configuration point with a default value and an ordered list of rules.

RuleEngine

Central registry. Stores variables and evaluates them against a context dict.

Usage

Evaluating a single variable

value = engine.evaluate("retriever_provider", context)

Evaluating multiple variables at once

Pass a list of names to get a dict back:

results = engine.evaluate(["retriever_provider", "model_name", "environment"], context)
# → {"retriever_provider": "aws", "model_name": "gpt-3.5-turbo", "environment": "dev"}

Multiple conditions (AND logic)

All conditions in a rule must match for the rule to fire:

rule = Rule(
    name="aws_for_application",
    conditions=[
        Condition(field="system_type", operator="eq",         value="application"),
        Condition(field="token_hash",  operator="starts_with", value="xpto"),
        Condition(field="token_hash",  operator="ends_with",   value="abc123"),
    ],
    result="aws",
    priority=1,
)

Priority

Rules are sorted by descending priority on register(). When two rules could match, the one with the highest priority wins:

rule_high = Rule(name="r1", conditions=[...], result="aws",    priority=2)
rule_low  = Rule(name="r2", conditions=[...], result="bridge", priority=1)
# rule_high is evaluated first

Serialization — export and reload

Variables can be serialized to plain dicts (e.g. to store in a database or config file) and reloaded:

payload = engine.export_variables()   # list[dict]
# ... save to DB / JSON file ...

new_engine = RuleEngine()
new_engine.load_variables(payload)

Complete example

from rudy_ab_test import RuleEngine, Variable, Rule, Condition

engine = RuleEngine()

# --- retriever_provider ---
engine.register(Variable(
    name="retriever_provider",
    default="bridge",
    rules=[
        Rule(
            name="aws_for_privileged_users",
            conditions=[
                Condition(field="system_type", operator="eq", value="user"),
                Condition(field="function",    operator="in", value=["SPECIALIST_1", "SUPERINTENDENT"]),
            ],
            result="aws",
            priority=2,
        ),
        Rule(
            name="aws_for_verified_apps",
            conditions=[
                Condition(field="system_type", operator="eq",          value="application"),
                Condition(field="token_hash",  operator="starts_with", value="xpto"),
                Condition(field="token_hash",  operator="ends_with",   value="abc123"),
            ],
            result="aws",
            priority=1,
        ),
    ],
))

# --- model_name ---
engine.register(Variable(
    name="model_name",
    default="gpt-3.5-turbo",
    rules=[
        Rule(name="prod_model", conditions=[Condition(field="environment", operator="eq", value="prod")], result="gpt-4",         priority=1),
        Rule(name="dev_model",  conditions=[Condition(field="environment", operator="eq", value="dev")],  result="gpt-3.5-turbo", priority=2),
    ],
))

# Evaluate a privileged user
user_ctx = {"system_type": "user", "function": "SPECIALIST_1", "environment": "dev"}
print(engine.evaluate(["retriever_provider", "model_name"], user_ctx))
# → {"retriever_provider": "aws", "model_name": "gpt-3.5-turbo"}

# Evaluate a standard user
anon_ctx = {"system_type": "user", "function": "JUNIOR_ANALIST", "environment": "dev"}
print(engine.evaluate(["retriever_provider", "model_name"], anon_ctx))
# → {"retriever_provider": "bridge", "model_name": "gpt-3.5-turbo"}

Requirements

• Python ≥ 3.10

License

MIT