Skip to content

Persistence

Prerequisites: Concepts (Rule) and Sessions (RuleService).

FluxRules rules can be persisted to a database. Use persistence when you need to:

  1. Store rules permanently - survive application restarts
  2. Share rules across instances - load the same ruleset in multiple processes
  3. Audit trail - track rule evaluation for compliance

Quick start: Store and load rules

from fluxrules import Rule
from fluxrules.engine.phreak import PhreakEngine

# Create rules
rules = [
    Rule(
        name="high_value_transaction",
        condition_dsl={"type": "condition", "field": "amount", "op": ">", "value": 5000},
        action="require_approval",
    ),
]

# Load and evaluate with engine
engine = PhreakEngine()
engine.load_rules(rules)
result = engine.evaluate({"amount": 6000})
print(f"Matched rules: {result.fired_rules}")

Persist flag

Every rule has a persist flag (default: False):

  • persist=False (default) - Constructing the rule does no I/O. The rule is in-memory only and gets a locally generated ID.
  • persist=True - The rule is written to the database during construction and adopts the database-assigned ID.

Constructing a Rule is a pure operation by default, so importing a module that defines rules never opens a database connection. To store a rule you built in memory, call Rule.save():

rule = Rule(
    name="production_rule",
    condition_dsl={"type": "condition", "field": "x", "op": ">", "value": 10},
    action="act",
)
rule.save()  # now in the database, rule.id is the stored ID

save() is an insert, not an upsert: calling it twice stores the rule twice and does not update an existing row. Use the repository or service layer when you need update semantics.

# Persisted rule (stored in database)
rule1 = Rule(
    name="production_rule",
    condition_dsl={"type": "condition", "field": "x", "op": ">", "value": 10},
    action="act",
    persist=True,  # Database storage
)

# Temporary rule (in-memory only)
rule2 = Rule(
    name="test_rule",
    condition_dsl={"type": "condition", "field": "y", "op": "==", "value": "test"},
    action="test_act",
    persist=False,  # No database storage
)

Configuration

Set the database environment during initialization:

from fluxrules.initialization import initialize_persistence

# Development (SQLite, default)
initialize_persistence(env="dev")

# Production (PostgreSQL)
initialize_persistence(env="prod", db_url="postgresql://user:pass@db.example.com/fluxrules")

# Custom URL
initialize_persistence(db_url="postgresql://localhost/my_rules")

Using environment variables:

export FLUXRULES_ENV=prod
export FLUXRULES_DB_URL=postgresql://user:pass@db.example.com/fluxrules

FluxRules auto-initializes the database on first use if not called explicitly.

Audit trail

Every rule evaluation produces an audit trail automatically:

from fluxrules import Rule
from fluxrules.engine.phreak import PhreakEngine

# Setup engine and rules
rule = Rule(name="test", condition_dsl={"type": "condition", "field": "amount", "op": ">", "value": 5000}, action="flag")
engine = PhreakEngine()
engine.load_rules([rule])

fact = {"amount": 6000}
result = engine.evaluate(fact)

# Audit trail built-in to result
print(f"Matched: {result.fired_rules}")      # Rule IDs that matched
print(f"Actions: {result.actions}")          # Actions executed
print(f"Time: {result.latency_ms}ms")        # Execution time

# Log for audit
import json
audit_record = {
    "fact_id": fact.get("id"),
    "matched_rules": result.fired_rules,
    "actions": result.actions,
    "latency_ms": result.latency_ms,
}
print(json.dumps(audit_record))

Persist audit records to your logging backend as needed.

Supported databases

  • SQLite (default, development, single-file)
  • PostgreSQL (production, scales to 100K+ rules)
  • SQL Server, MySQL, and any SQLAlchemy-supported dialect — via the ServiceFactory path below

Using a different SQL provider (Postgres, SQL Server, etc.)

By default persist=True writes through the DBConnectionManager path, which is configured via initialize_persistence() or the FLUXRULES_DB_URL / FLUXRULES_ENV environment variables.

In real applications you typically configure a connection once at startup (e.g. in a FastAPI lifespan handler). When you register a persistence adapter with ServiceFactory, persist=True automatically delegates to that adapter instead of the default SQLite path — no other changes needed.

FastAPI + PostgreSQL example

# python skip
from contextlib import asynccontextmanager
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from fastapi import FastAPI

from fluxrules.config import PersistenceConfig, RuntimeConfig
from fluxrules.di import ServiceFactory
from fluxrules import Rule

DATABASE_URL = "postgresql://user:pass@db.example.com/fluxrules"
engine = create_engine(DATABASE_URL, pool_size=10, max_overflow=20)
SessionLocal = sessionmaker(bind=engine)

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Register the Postgres-backed adapter once at startup.
    ServiceFactory.create_persistence_adapter(
        session_provider=SessionLocal,
        config=RuntimeConfig(persistence=PersistenceConfig(backend="postgres")),
    )
    yield
    ServiceFactory.reset()

app = FastAPI(lifespan=lifespan)

# Anywhere in your app — persist=True now writes to Postgres, not SQLite.
rule = Rule(
    name="kyc_verified",
    domain="compliance",
    condition_dsl={"type": "condition", "field": "kyc_status", "op": "==", "value": "verified"},
    action="proceed",
    persist=True,
)

The same pattern works for SQL Server, MySQL, or any dialect SQLAlchemy supports — just change the DATABASE_URL scheme (e.g. mssql+pyodbc://...).

Priority order

When persist=True is used, fluxrules checks in this order:

  1. ServiceFactory adapter — if one is registered and it is not the in-memory fallback, it is used. This covers Postgres, SQL Server, and any custom adapter.
  2. DBConnectionManager — the initialize_persistence() / env-var path. Used when no ServiceFactory adapter is registered (the common case for standalone scripts and simple deployments).
  3. Fallback — if neither is reachable, a locally generated ID is assigned and a DEBUG-level log entry is emitted. The rule is still usable in-memory.

Note — table creation: ServiceFactory does not create database tables. When using the ServiceFactory path you are responsible for running schema migrations before the application starts (e.g. via Alembic). The initialize_persistence() helper creates tables automatically and is the right choice when you want a zero-config startup with SQLite or a simple PostgreSQL setup.

Next Steps