Rule Cascade in Python
Paket: pip install rule-cascade (Python 3.11+; Import rule_cascade). Bündel kommen von
rcas compile --all nach build/rules/. Werte YAML niemals zur Laufzeit aus und kopiere niemals eine Regel in den Code.
Einmal laden, pro Anfrage auswerten
import json
from rule_cascade import RuleSet, RequestError
# Beim Start (Modulebene, FastAPI-Lifespan, Django AppConfig.ready). Ein fehlerhaftes Bündel löst einen Fehler aus: Nicht ausliefern.
with open("build/rules/<ruleset-id>.bundle.json", encoding="utf-8") as f:
rules = RuleSet.from_bundle(json.load(f))
Den Dekorator bevorzugen (ab 1.0.0-alpha.7): Er erstellt die Anfrage, erzwingt die Regeln vor der Ausführung der Funktion und löst RuleViolation aus, dessen problem() die 422-Antwort unten ist.
from rule_cascade import enforce_rules, current_evaluation
from rule_cascade.fastapi import install # oder rule_cascade.flask.init_app(app), oder die Django-Middleware
install(app) # RuleViolation -> 422, RequestError -> 400
@app.post("/orders", status_code=201)
@enforce_rules(rules, entity="Order", operation="create", data="order", actor="user") # unter dem Routen-Dekorator
def create_order(order: Order, user: User = Depends(current_user)):
evaluation = current_evaluation() # das erlaubte Ergebnis: Warnungen, berechnete Werte, Befehle
rules.enforce(request, "server") macht dasselbe ohne Dekorator. Dahinter gibt
rules.evaluate(request, "server") ein Dictionary zurück: decision (allow oder deny),
findings, effects, commands. Es löst RequestError für eine fehlerhafte Anfrage aus (Antwort 400). Jeder zustandsändernde Endpunkt folgt vier Schritten:
- Auf dem Server auswerten mit
{"entity", "operation", "data", "original", "actor", "resolutions"};actorkommt aus der Authentifizierung ({"id", "roles"}), niemals aus einem Feld, das der Aufrufer sendet. - Bei deny ablehnen mit 422 Problem-Details (
application/problem+json,type: "urn:rule-cascade:rule-violation"), die das Ergebnis enthalten, sodass Clientsfindings[].fieldsauf Eingaben abbilden können. - Persistieren, nachdem
compute-Effekte angewendet wurden (effect["type"] == "value"): Der Wert des Servers wird gespeichert. result["commands"]ausführen nach dem Commit, einmal proidempotencyKey(eine Outbox-Tabelle oder eine Aufgabenwarteschlange mit Deduplizierung).
Framework-Platzierung: FastAPI eine Abhängigkeit (Depends(get_rules)) und einen Exception-Handler, der eine RuleViolation in die 422 umwandelt; Django eine Service-Funktion, die von jedem View oder Serializer-validate() der Entität aufgerufen wird, plus Middleware oder DRF-Exception-Handler für die 422; Flask ein Erweiterungsobjekt, das in der App-Factory erstellt wird. Celery- und Batch-Jobs werten vor jedem Schreibzugriff auf die gleiche Weise aus.
RuleSetHolder(load_bundle, interval=300) (oder cron="0 * * * *") lädt Bündel im Hintergrund neu und behält die letzten gültigen Regeln, wenn ein Neuladen fehlschlägt.
Viele Datensätze
Starte für Dateien und Warteschlangen einmal rcas engine und sende JSON-Lines-Anfragen hindurch, oder rufe rules.evaluate in einer Schleife auf: Lade das Bündel in beiden Fällen nur einmal. Siehe https://rulescascade.com/examples/batch/.
Bestehende Validierung ersetzen
rcas analyze --module <module> listet Pydantic Field(le=...) und Validatoren, Marshmallow-Schemas, Django-Validatoren, clean()-Methoden und Berechtigungen, handgeschriebene raise-Prüfungen, benannte Limits (MAX_QTY = 10), Vergleiche mit Zahlen und match-Zweige auf. Ein Kandidat mit drift bedeutet, dass eine andere Sprache des Projekts (oft das Frontend) denselben Namen mit einem anderen Wert entscheidet: Ein Regelsatz muss ihn für beide entscheiden. Behalte Typ- und Formvalidierung (Pydantic-Modelle) dort, wo sie ist; verschiebe Geschäftsentscheidungen in Regeln. Füge einen Paritätstest hinzu, der die alte Prüfung und die Engine mit denselben Eingaben ausführt, und lösche die alte Prüfung erst, wenn sie übereinstimmen.
Tests
Prüfe decision und Finding-codes, nicht Nachrichtentexte. Die Regeln selbst werden durch ihre goldenen Tests (rcas check) getestet. Führe in CI rcas check und rcas compile --all vor pytest aus.
Referenz: https://rulescascade.com/usage/python/