pylitmus 1.2.0

A high-performance rules engine for Python - evaluate data against configurable rules and get clear verdicts

pylitmus

A high-performance rules engine for Python. Like a litmus test for your data - evaluate against configurable rules and get clear verdicts.

Features

• YAML/JSON rule definitions - Business-friendly rule configuration
• Hot-reload - Rules can be updated without restart
• Multiple storage backends - Memory, database, file
• Caching - Redis and in-memory caching support
• Multiple scoring strategies - Sum, weighted, max
• Flask integration - Easy integration with Flask apps
• 18 built-in operators - Comparison, collection, string, null, temporal
• Extensible - Custom evaluators and strategies

Installation

pip install pylitmus

# With database support
pip install pylitmus[database]

# With Redis caching
pip install pylitmus[redis]

# With Flask integration
pip install pylitmus[flask]

# All extras
pip install pylitmus[all]

Quick Start

from pylitmus import create_engine, Rule, Severity

# Create engine with inline rules
engine = create_engine(rules=[
    Rule(
        code='AMT_001',
        name='High Amount',
        description='Flag high amounts',
        category='AMOUNT',
        severity=Severity.HIGH,
        score=60,
        enabled=True,
        conditions={'field': 'amount', 'operator': 'greater_than', 'value': 5000}
    )
])

# Evaluate data
result = engine.evaluate({'amount': 6000})

print(f"Score: {result.total_score}")      # 60
print(f"Decision: {result.decision}")       # FLAG
print(f"Triggered: {[r.rule_code for r in result.triggered_rules]}")  # ['AMT_001']

YAML Rules

Define rules in YAML files for easy management:

# rules.yaml
rules:
  - code: "AMT_001"
    name: "High Amount"
    description: "Flag transactions over $5000"
    category: "AMOUNT"
    severity: "HIGH"
    score: 60
    enabled: true
    conditions:
      field: "amount"
      operator: "greater_than"
      value: 5000

  - code: "RISK_001"
    name: "High Risk Country"
    description: "Flag transactions from high-risk countries"
    category: "RISK"
    severity: "CRITICAL"
    score: 80
    enabled: true
    conditions:
      field: "country"
      operator: "in"
      value: ["XX", "YY", "ZZ"]

Load rules from file:

engine = create_engine(
    storage_backend='file',
    rules_file='rules.yaml'
)

Composite Conditions

Combine conditions with AND/OR logic:

conditions:
  all:  # AND
    - field: "amount"
      operator: "greater_than"
      value: 1000
    - any:  # OR
        - field: "is_new_customer"
          operator: "equals"
          value: true
        - field: "country"
          operator: "in"
          value: ["NG", "KE", "GH"]

Or use the alternative format:

conditions:
  type: "AND"
  conditions:
    - field: "amount"
      operator: "greater_than"
      value: 1000
    - field: "is_international"
      operator: "equals"
      value: true

Available Operators

Category	Operators
Comparison	equals, not_equals, greater_than, greater_than_or_equal, less_than, less_than_or_equal, between
Collection	in, not_in, contains, not_contains
String	starts_with, ends_with, matches_regex
Null	is_null, is_not_null
Temporal	within_days, before, after

Scoring Strategies

Sum Strategy (Default)

Adds up all triggered rule scores, capped at 100.

engine = create_engine(scoring_strategy='sum')

Weighted Strategy

Uses severity-based weights (LOW=1, MEDIUM=2, HIGH=3, CRITICAL=4).

engine = create_engine(scoring_strategy='weighted')

Max Strategy

Takes the highest score from triggered rules.

engine = create_engine(scoring_strategy='max')

Decision Thresholds

Customize decision boundaries:

engine = create_engine(
    decision_thresholds={
        'approve': 30,  # Score < 30 = APPROVE
        'review': 70    # Score 30-70 = REVIEW, >= 70 = FLAG
    }
)

Storage Backends

In-Memory

engine = create_engine(storage_backend='memory', rules=[...])

File-Based

engine = create_engine(
    storage_backend='file',
    rules_file='rules.yaml'  # or rules.json
)

Database

engine = create_engine(
    storage_backend='database',
    database_url='postgresql://localhost/mydb'
)

Caching

Memory Cache

engine = create_engine(
    cache_backend='memory',
    cache_ttl=300  # 5 minutes
)

Redis Cache

engine = create_engine(
    cache_backend='redis',
    cache_url='redis://localhost:6379/0',
    cache_ttl=600
)

No Cache

engine = create_engine(cache_backend='none')

Flask Integration

from flask import Flask
from pylitmus.integrations.flask import CmapRulesEngine, get_engine

app = Flask(__name__)
app.config['CMAP_RULES_FILE'] = 'rules.yaml'

rules_engine = CmapRulesEngine(app)

@app.route('/evaluate', methods=['POST'])
def evaluate():
    data = request.json
    engine = get_engine()
    result = engine.evaluate(data)
    return {
        'score': result.total_score,
        'decision': result.decision,
        'triggered_rules': [r.rule_code for r in result.triggered_rules]
    }

Nested Field Access

Access nested data using dot notation:

data = {
    'transaction': {
        'amount': 6000,
        'merchant': {
            'category': 'electronics'
        }
    }
}

# Rule condition
conditions:
  field: "transaction.merchant.category"
  operator: "equals"
  value: "electronics"

Pattern Matching

Advanced pattern matching capabilities:

from pylitmus import EnhancedPatternEngine

pattern_engine = EnhancedPatternEngine()

# Regex matching
pattern_engine.add_pattern('email', r'^[\w.-]+@[\w.-]+\.\w+$', 'regex')

# Fuzzy matching
pattern_engine.add_pattern('name', 'John Smith', 'fuzzy', threshold=0.8)

# Range matching
pattern_engine.add_pattern('age', {'min': 18, 'max': 65}, 'range')

# Check matches
result = pattern_engine.match_all({
    'email': '[e-mail]',
    'name': 'Jon Smith',
    'age': 25
})

API Reference

create_engine()

def create_engine(
    storage_backend: str = 'memory',
    database_url: str = None,
    rules_file: str = None,
    rules: List[Rule] = None,
    repository: RuleRepository = None,
    cache_backend: str = 'memory',
    cache_url: str = None,
    cache_ttl: int = 300,
    scoring_strategy: str = 'sum',
    decision_thresholds: Dict[str, int] = None,
) -> RuleEngine

RuleEngine.evaluate()

def evaluate(
    self,
    data: Dict[str, Any],
    context: Dict[str, Any] = None,
    filters: Dict[str, Any] = None
) -> AssessmentResult

AssessmentResult

@dataclass
class AssessmentResult:
    total_score: int           # Total calculated score
    decision: str              # APPROVE, REVIEW, or FLAG
    triggered_rules: List[RuleResult]  # Rules that matched
    processing_time_ms: float  # Processing time in ms

Full Example

from pylitmus import (
    create_engine,
    Rule,
    Severity,
    InMemoryRuleRepository,
    WeightedStrategy,
)

# Define rules
rules = [
    Rule(
        code='AMT_HIGH',
        name='High Amount',
        description='Flag high-value transactions',
        category='AMOUNT',
        severity=Severity.HIGH,
        score=60,
        enabled=True,
        conditions={'field': 'amount', 'operator': 'greater_than', 'value': 5000}
    ),
    Rule(
        code='NEW_CUSTOMER',
        name='New Customer',
        description='Flag new customer transactions',
        category='CUSTOMER',
        severity=Severity.MEDIUM,
        score=30,
        enabled=True,
        conditions={'field': 'is_new_customer', 'operator': 'equals', 'value': True}
    ),
    Rule(
        code='INTL_TXN',
        name='International Transaction',
        description='Flag international transactions',
        category='GEOGRAPHY',
        severity=Severity.LOW,
        score=20,
        enabled=True,
        conditions={'field': 'is_international', 'operator': 'equals', 'value': True}
    ),
]

# Create engine with weighted scoring
engine = create_engine(
    rules=rules,
    scoring_strategy='weighted',
    decision_thresholds={'approve': 25, 'review': 60}
)

# Evaluate transaction
result = engine.evaluate({
    'amount': 6000,
    'is_new_customer': True,
    'is_international': False
})

print(f"Total Score: {result.total_score}")
print(f"Decision: {result.decision}")
print(f"Triggered Rules: {[r.rule_code for r in result.triggered_rules]}")
print(f"Processing Time: {result.processing_time_ms:.2f}ms")

Documentation

• Quick Start Guide
• Rule Format
• API Reference
• Flask Integration
• Examples

License

MIT License - see LICENSE for details.