Skip to content

Logic Patterns

LogicBank Patterns - The Hitchhiker's Guide

This document contains general patterns for working with LogicBank rules. These patterns apply to ALL rule types (deterministic and probabilistic).

For specific rule APIs, see: - Eval-logic_bank_api.md - Deterministic rules (sum, count, formula, constraint, etc.) - Eval-probabilistic_logic.md - Probabilistic rules (AI value computation)

============================================================================= PATTERN 1: Event Handler Signature =============================================================================

ALL event handlers (early_row_event, commit_row_event, row_event) receive THREE parameters.

✅ REQUIRED SIGNATURE: 

def my_handler(row: models.MyTable, old_row: models.MyTable, logic_row: LogicRow):
    """
    Event handler signature - ALL THREE PARAMETERS REQUIRED

    Args:
        row: Current state of the row (with changes)
        old_row: Previous state before changes (for detecting what changed)
        logic_row: LogicBank's wrapper with rule execution methods
    """
    logic_row.log(f"Processing {row.__class__.__name__}")
    # Your logic here

❌ WRONG: Trying to "get" logic_row 

def my_handler(row: models.MyTable):
    logic_row = LogicRow.get_logic_row(row)  # ❌ This method does NOT exist!

❌ WRONG: Missing parameters 

def my_handler(row: models.MyTable, logic_row: LogicRow):  # ❌ Missing old_row
    pass

REGISTRATION: 

# Option 1: Direct registration (LogicBank passes all three params)
Rule.early_row_event(on_class=models.MyTable, calling=my_handler)

# Option 2: Lambda wrapper (if you need to pass additional args)
Rule.early_row_event(
    on_class=models.MyTable,
    calling=lambda row, old_row, logic_row: my_handler(row, old_row, logic_row, extra_arg)
)

WHY THREE PARAMETERS: - row - Access current values, make changes - old_row - Detect what changed (if row.price != old_row.price) - logic_row - Access LogicBank methods (.log(), .new_logic_row(), .insert(), etc.)

============================================================================= PATTERN 2: Logging with logic_row.log() =============================================================================

ALWAYS use logic_row.log() for rule execution logging (not app_logger).

✅ CORRECT: Use logic_row.log() 

def my_handler(row: models.Item, old_row, logic_row: LogicRow):
    logic_row.log(f"Processing Item - quantity={row.quantity}")

    if row.product.count_suppliers > 0:
        logic_row.log(f"Product has {row.product.count_suppliers} suppliers")
    else:
        logic_row.log("No suppliers available, using default price")

❌ WRONG: Using app_logger for rule logic 

def my_handler(row: models.Item, old_row, logic_row: LogicRow):
    app_logger.info(f"Item {row.id} - Product has suppliers")  # ❌ Wrong!

BENEFITS of logic_row.log(): - ✅ Automatic indentation showing rule cascade depth - ✅ Grouped with related logic execution in trace output - ✅ Visible in logic trace (helps debugging) - ✅ No need to import logging module - ✅ Shows execution context (which rule fired)

WHEN TO USE app_logger: - System startup messages - Configuration loading - Errors outside rule execution - Non-rule application logic

EXAMPLE OUTPUT: 

Logic Phase:        ROW LOGIC       (sqlalchemy before_flush)            - 2025-11-14 06:19:03,372 - logic_logger - INF
..Item[None] {Insert - client} Id: None, order_id: 1, product_id: 6, quantity: 10, unit_price: None, amount: None  row: 0x107e4a950  session: 0x107e4a8d0  ins_upd_dlt: ins - 2025-11-14 06:19:03,373 - logic_logger - INF
....Processing Item - quantity=10 - 2025-11-14 06:19:03,373 - logic_logger - INF
....Product has 2 suppliers - 2025-11-14 06:19:03,374 - logic_logger - INF
......Creating SysSupplierReq for AI selection - 2025-11-14 06:19:03,375 - logic_logger - INF

Note the indentation (dots) showing call depth!

============================================================================= PATTERN 3: Request Pattern (ROP) - Integration Services =============================================================================

📚 Full Documentation: See Eval-RequestObjectPattern.md for comprehensive guide

Quick Definition: The Request Pattern is a table design for integration services with automatic audit: - Request fields = user input (e.g., product_id, hs_code_id, value_amount) - Response fields = computed output (e.g., chosen_supplier_id, duty_amount, reason) - early_row_event = performs integration service (AI, external API, calculations)

When to Use: ✅ AI decisions (supplier selection, pricing, routing) ✅ External API calls (payment gateways, shipping carriers) ✅ Messaging/Email services (Kafka, SMTP with templating) ✅ Complex calculations requiring audit (customs, tax, compliance)

Recognition Signal in Prompts: - "calculate/determine/select [X] when [Y] is given" - Integration service needed - Compliance/audit domain

Architecture: 

User/API → Insert with request fields only
         ↓
   early_row_event fires (integration service logic)
         ↓
   Populates response fields
         ↓
   Automatic audit trail persisted

Benefits: - Single source of truth (works from API, Admin UI, batch, tests) - Automatic audit trail for compliance - Governed by deterministic rules - Testable without API/HTTP - No duplication

Anti-Pattern: ❌ Fat API services with business logic (bypasses rules engine, no audit, duplication)

Technical Implementation with new_logic_row():

Use new_logic_row() to create Request Pattern instances in event handlers.

✅ CORRECT: Pass MODEL CLASS to new_logic_row 

def create_audit_trail(row: models.Order, old_row, logic_row: LogicRow):
    """Create audit request object using Request Pattern"""

    # Step 1: Create request object (pass CLASS not instance)
    request_logic_row = logic_row.new_logic_row(models.OrderAuditReq)

    # Step 2: Get the instance from .row property
    request = request_logic_row.row

    # Step 3: Set attributes on the instance
    request.order_id = row.id
    request.customer_id = row.customer_id
    request.action = "order_created"
    request.request_data = {"amount": float(row.amount_total)}

    # Step 4: Insert using logic_row (triggers any events on request table)
    request_logic_row.insert(reason="Order audit trail")

    # Step 5: Access results if needed
    logic_row.log(f"Audit created with ID {request.id}")

❌ WRONG: Creating instance first 

def create_audit_trail(row: models.Order, old_row, logic_row: LogicRow):
    # ❌ Don't create instance yourself
    request = models.OrderAuditReq()

    # ❌ This will fail with TypeError: object is not callable
    request_logic_row = logic_row.new_logic_row(request)

THE METHOD SIGNATURE: 

logic_row.new_logic_row(a_class: type) -> LogicRow

WHAT IT RETURNS: - Returns a LogicRow wrapper (not the instance directly) - Access instance via .row property - Use returned logic_row for .insert(), .link(), etc.

WHY THIS PATTERN: - LogicBank needs to track the new row in the session - Enables rule execution on the new row - Maintains parent-child relationships - Supports cascading logic across related objects

COMMON USE CASES: 1. Audit trails - Track who did what when 2. Workflows - Create approval requests, notifications 3. AI integration - Create request objects for AI to populate 4. Derived objects - Generate summary records, reports

============================================================================= PATTERN 4: Rule API Syntax Reference =============================================================================

Always consult Eval-logic_bank_api.md for complete API details.

COMMON PARAMETERS BY RULE TYPE:

Rule.sum() - NO 'calling' parameter Rule.sum(derive: Column, as_sum_of: any, where: Callable = None, insert_parent: bool = False) ✅ Use 'where' for filtering ❌ NO 'calling' parameter

Rule.count() - NO 'calling' parameter Rule.count(derive: Column, as_count_of: type, where: Callable = None, insert_parent: bool = False) ✅ Use 'where' for filtering ❌ NO 'calling' parameter

Rule.formula() - HAS 'calling' parameter (for functions only) Rule.formula(derive: Column, as_expression: Callable = None, calling: Callable = None, no_prune: bool = False) ✅ Use 'as_expression' for simple expressions ✅ Use 'calling' for complex functions (must be callable, not bool) ❌ Never use calling=False or calling=True

Rule.constraint() - HAS 'calling' parameter (for functions only) Rule.constraint(validate: type, as_condition: Callable = None, calling: Callable = None, error_msg: str = "") ✅ Use 'as_condition' for simple lambda conditions ✅ Use 'calling' for complex validation functions ❌ Never use calling=False or calling=True

🚨 PITFALL — LogicBank dependency scanner confusion in calling= function bodies: LogicBank tokenizes the function body to discover attribute dependencies. If the function body contains a pattern like session.query(...).filter(... == row.project_id).first(), the scanner reads project_id).first as the attribute name and raises: LBActivateException: ['Charge.project_id).first: constraint']

Fix: extract row.* reads to local variables BEFORE any method-chained expressions: 

# ❌ WRONG — scanner sees 'project_id).first' as an attribute name
def check_funding(row, old_row, logic_row):
    project = session.query(models.Project).filter(
        models.Project.id == row.project_id).first()   # row.project_id inside chained call

# ✅ CORRECT — extract to local variable first
def check_funding(row, old_row, logic_row):
    project_id = row.project_id                        # scanner sees clean 'project_id'
    project = session.query(models.Project).filter(
        models.Project.id == project_id).first()       # local var, not row.attr

Rule.copy() - NO 'calling' parameter Rule.copy(derive: Column, from_parent: any)

Rule.parent_check() - NO 'calling' parameter Rule.parent_check(validate: type, error_msg: str = "")

EXAMPLES:

✅ CORRECT: Rule.count with where 

Rule.count(
    derive=models.Customer.unshipped_order_count,
    as_count_of=models.Order,
    where=lambda row: row.date_shipped is None
)

❌ WRONG: Rule.count with calling 

Rule.count(
    derive=models.Customer.unshipped_order_count,
    as_count_of=models.Order,
    calling=lambda row: row.date_shipped is None  # ❌ 'calling' not valid!
)

✅ CORRECT: Rule.formula with conditional 

Rule.formula(
    derive=models.Item.unit_price,
    as_expression=lambda row: (
        row.product.unit_price if row.product.count_suppliers == 0
        else row.unit_price  # Preserve value from event
    )
)

❌ WRONG: Rule.formula with calling=False 

Rule.formula(
    derive=models.Item.unit_price,
    calling=False  # ❌ calling must be callable or omitted!
)

============================================================================= PATTERN 5: Common Anti-Patterns (What NOT to Do) =============================================================================

❌ DON'T: Try to "get" logic_row 

# This method does NOT exist
logic_row = LogicRow.get_logic_row(row)

❌ DON'T: Use app_logger in rule code 

# Use logic_row.log() instead
app_logger.info("Processing item")

❌ DON'T: Create instances before new_logic_row 

# Pass CLASS to new_logic_row, not instance
request = models.AuditReq()
logic_row.new_logic_row(request)  # ❌ TypeError!

❌ DON'T: Use wrong parameters for rules 

# Rule.count/sum/copy don't have 'calling'
Rule.count(derive=..., as_count_of=..., calling=...)  # ❌ Invalid!

❌ DON'T: Use calling with boolean values 

# calling must be a function, not bool
Rule.formula(derive=..., calling=False)  # ❌ Invalid!

❌ DON'T: Forget to copy AI results to target 

# AI populates request table, you must copy to target
request_logic_row.insert()
# ❌ Missing: row.unit_price = request.chosen_unit_price

❌ DON'T: Skip event handler parameters 

# Event handlers need all THREE parameters
def my_handler(row):  # ❌ Missing old_row and logic_row
    pass

============================================================================= PATTERN 6: Type Handling for Database Fields =============================================================================

When setting values in event handlers or custom code, use correct Python types for database columns.

✅ FOREIGN KEY (ID) FIELDS - Use int 

def my_handler(row, old_row, logic_row: LogicRow):
    # ✅ CORRECT: Foreign keys must be int for SQLite
    row.customer_id = 123  # int
    row.supplier_id = int(some_value)  # Ensure it's int

❌ WRONG: Using Decimal for foreign keys 

row.customer_id = Decimal('123')  # ❌ SQLite error: Decimal not supported for INTEGER FK

✅ MONETARY FIELDS - Use Decimal for precision 

from decimal import Decimal

def my_handler(row, old_row, logic_row: LogicRow):
    # ✅ CORRECT: Monetary values as Decimal
    row.unit_price = Decimal('19.99')
    row.amount = Decimal(str(quantity * price))  # Convert via string for precision

✅ PATTERN SUMMARY: 

# Foreign keys and IDs
if '_id' in field_name or field_name.endswith('_id'):
    value = int(value)  # Must be int for SQLite INTEGER columns

# Monetary fields  
elif '_price' in field_name or '_cost' in field_name or '_amount' in field_name:
    value = Decimal(str(value))  # Use Decimal for precision

# Other numerics
else:
    value = float(value) or int(value)  # Based on column type

WHY THIS MATTERS: - SQLite INTEGER columns (foreign keys) don't support Decimal type - Monetary calculations need Decimal to avoid floating-point errors - Type mismatches cause "type not supported" database errors

COMMON ERRORS: 

# ❌ WRONG: Decimal for foreign key
row.order_id = Decimal('42')  # Database error!

# ❌ WRONG: Float for money (precision loss)
row.unit_price = 19.99  # May lose precision in calculations

# ✅ CORRECT: Proper types
row.order_id = 42  # int for FK
row.unit_price = Decimal('19.99')  # Decimal for money

============================================================================= PATTERN 7: Testing and Debugging Patterns =============================================================================

✅ USE logic_row.log() EXTENSIVELY during development 

def my_handler(row, old_row, logic_row: LogicRow):
    logic_row.log("=== Starting my_handler ===")
    logic_row.log(f"Row state: quantity={row.quantity}, price={row.unit_price}")

    if row.quantity != old_row.quantity:
        logic_row.log(f"Quantity changed: {old_row.quantity} -> {row.quantity}")

    # ... your logic

    logic_row.log(f"=== Completed my_handler, result={row.amount} ===")

✅ CHECK old_row to detect changes 

def update_handler(row, old_row, logic_row: LogicRow):
    if row.status != old_row.status:
        logic_row.log(f"Status changed: {old_row.status} -> {row.status}")
        # Take action on status change

✅ USE logic_row.is_inserted(), is_updated(), is_deleted() 

def audit_handler(row, old_row, logic_row: LogicRow):
    if logic_row.is_inserted():
        logic_row.log("New row created")
    elif logic_row.is_updated():
        logic_row.log("Row updated")
    elif logic_row.is_deleted():
        logic_row.log("Row deleted")

✅ TRACE rule execution with PYTHONPATH 

# Enable verbose logic logging
export APILOGICSERVER_VERBOSE=True
python api_logic_server_run.py

============================================================================= PATTERN 8: Rules vs Events — Rules Are ALWAYS Preferred =============================================================================

ALWAYS use declarative rules (Rule.formula, Rule.sum, Rule.count, Rule.constraint) in preference to events. Events are for side effects only.

LogicBank formulas order themselves automatically — you do not need an event to control ordering. If a value is derived from other attributes or child rows, use a rule.

Value source Use
Derived from row attributes or child rows Rule.formula, Rule.sum, Rule.count
Looked up from a FK parent (external to this row) Rule.early_row_event — only when no rule can express it
External module returns multiple output fields (AI response, external API, Request Pattern) Rule.early_row_event — handler sets all response fields; downstream Rule.formula rules consume them
Pure side effect (Kafka, email, audit, insert child rows) Rule.row_event, Rule.after_flush_row_event, Rule.commit_row_event

External module exception — when early_row_event legitimately sets multiple row attributes: When an event calls an opaque external computation (AI model, third-party API, Kafka-reply handler via the Request Pattern), the handler writes back several response fields (e.g. matched_project_id, confidence, reason, chosen_unit_price). This is correct and by design: the external call is a single atomic operation whose outputs are all response columns on a Sys* request table. Downstream Rule.formula rules then consume those output columns as inputs. This is NOT the same as a Rule.formula function setting multiple row. attributes as side-effects — that remains wrong (see logic_bank_api.md "ONE VALUE PER FORMULA"). The distinction: - early_row_event handler setting multiple fields = ✅ external module pattern (opaque computation, all fields are outputs) - Rule.formula calling function setting multiple fields = ❌ side-effect anti-pattern (only the derive= column is tracked)

❌ WRONG — using an event to derive a value that Rule.formula/count could express: 

def _evaluate(row, old_row, logic_row):
    row.clvs_eligible = 1 if not any(c.is_prohibited for c in row.ShipmentCommodityList) else 0
Rule.commit_row_event(on_class=models.Shipment, calling=_evaluate)
This fires once on commit, never re-fires when child rows change, and bypasses all LogicBank dependency tracking. It is always wrong for derived attributes.

✅ CORRECT — Rule.count + Rule.formula: reactive, dependency-tracked, re-fires automatically: 

Rule.count(derive=models.Shipment.prohibited_count, as_count_of=models.ShipmentCommodity,
           where=lambda row: row.is_prohibited == 1)
Rule.formula(derive=models.Shipment.clvs_eligible,
             as_expression=lambda row: 1 if row.prohibited_count == 0 else 0)

Valid Rule event methods (side effects only):

Use case Method
FK lookup before formulas run (no rule equivalent) Rule.early_row_event
Append child rows during parent insert Rule.row_event
Kafka publish / side-effect after flush Rule.after_flush_row_event
Post-commit finalization Rule.commit_row_event

Event firing order — two rules:

Scope Order
Within a single declare_logic() Declaration order is guaranteed — LogicBank appends to a plain list
Across files loaded by auto_discovery Non-deterministic — os.walk() file order is not specified

Consequence: if two early_row_event registrations on the same class live in different files, you cannot rely on which fires first. Fix: declare both in the same declare_logic(), in the required order. See Eval-allocate.md Variant C for the concrete Allocate + AI case.

⚠️ Note: Allocate is internally a subclass of EarlyRowEvent — it competes in the same ordering list as regular early_row_events on the same provider class.

Why it matters: Formulas and sums run after events. If your event sets a value that a formula consumes, using the wrong event type means the formula runs on the old value — silently, no error, wrong results.

# ✅ Sets unit_price BEFORE Item.amount formula runs
Rule.early_row_event(on_class=models.Item, calling=set_unit_price)

# ✅ Sends to Kafka AFTER all rules and constraints complete
Rule.after_flush_row_event(on_class=models.Order, calling=kafka_producer.send_row_to_kafka, ...)

# ❌ Wrong for setting unit_price — formula already ran with old value
Rule.row_event(on_class=models.Item, calling=set_unit_price)

============================================================================= SUMMARY: Quick Reference =============================================================================

  1. Event handlers: def handler(row, old_row, logic_row) - ALL THREE
  2. Logging: Use logic_row.log() not app_logger
  3. Request Pattern: new_logic_row(ModelClass) returns LogicRow with .row
  4. Rule APIs: Check logic_bank_api.md for correct parameters
  5. Anti-patterns: No get_logic_row(), no calling=False, no app_logger in rules
  6. Type handling: int for FKs, Decimal for money
  7. Rules over events: derived values → Rule.formula/sum/count; events → side effects only
  8. Testing: logic_row.log(), check old_row, use is_inserted/updated/deleted
  9. Valid event methods: early_row_event, row_event, after_flush_row_event, commit_row_event

For rule-specific APIs and examples: - Deterministic rules → Eval-logic_bank_api.md - Probabilistic rules → Eval-probabilistic_logic.md

END OF GENERAL PATTERNS