YarlisAISolutions/rules-cascade-python

Rule Cascade en Python

Package : pip install rule-cascade (Python 3.11+ ; import rule_cascade). Les bundles sont générés par rcas compile --all dans build/rules/. N'évaluez jamais le YAML à l'exécution et ne copiez jamais une règle dans le code.

Charger une fois, évaluer par requête

import json
from rule_cascade import RuleSet, RequestError

# Au démarrage (niveau module, lifespan FastAPI, AppConfig.ready Django). Un bundle invalide lève une erreur : ne pas servir.
with open("build/rules/<ruleset-id>.bundle.json", encoding="utf-8") as f:
    rules = RuleSet.from_bundle(json.load(f))

Préférez le décorateur (depuis 1.0.0-alpha.7) : il construit la requête, applique les contraintes avant l'exécution de la fonction, et lève RuleViolation, dont problem() correspond à la réponse 422 ci-dessous.

from rule_cascade import enforce_rules, current_evaluation
from rule_cascade.fastapi import install            # ou rule_cascade.flask.init_app(app), ou le middleware Django

install(app)                                        # RuleViolation -> 422, RequestError -> 400

@app.post("/orders", status_code=201)
@enforce_rules(rules, entity="Order", operation="create", data="order", actor="user")   # sous le décorateur de route
def create_order(order: Order, user: User = Depends(current_user)):
    evaluation = current_evaluation()               # le résultat autorisé : avertissements, valeurs calculées, commandes

rules.enforce(request, "server") fait la même chose sans décorateur. En interne, rules.evaluate(request, "server") retourne un dictionnaire : decision (allow ou deny), findings, effects, commands. Il lève RequestError pour une requête malformée (réponse 400). Chaque point d'accès qui modifie l'état suit quatre étapes :

1. Évaluer sur le serveur avec {"entity", "operation", "data", "original", "actor", "resolutions"} ; actor provient de l'authentification ({"id", "roles"}), jamais d'un champ envoyé par l'appelant.
2. Refuser sur deny avec des détails de problème 422 (application/problem+json, type: "urn:rule-cascade:rule-violation") qui incluent le résultat, afin que les clients puissent mapper findings[].fields aux entrées.
3. Persister, après application des effets compute (effect["type"] == "value") : la valeur du serveur est stockée.
4. Exécuter result["commands"] après la validation, une fois par idempotencyKey (une table de publication, ou une file de tâches avec déduplication).

Placement dans les frameworks : FastAPI — une dépendance (Depends(get_rules)) et un gestionnaire d'exception qui transforme une RuleViolation en 422 ; Django — une fonction de service appelée par chaque vue ou validate() du serializer de l'entité, plus un middleware ou un gestionnaire d'exception DRF pour la 422 ; Flask — un objet d'extension créé dans la fabrique d'application. Celery et les jobs batch évaluent avant chaque écriture de la même manière.

RuleSetHolder(load_bundle, interval=300) (ou cron="0 * * * *") recharge les bundles en arrière-plan et conserve les dernières règles valides si un rechargement échoue.

Nombreux enregistrements

Pour les fichiers et les files d'attente, démarrez rcas engine une fois et diffusez les requêtes JSON Lines via celui-ci, ou appelez rules.evaluate dans une boucle : chargez le bundle une fois dans les deux cas. Voir https://rulescascade.com/examples/batch/.

Remplacer une validation existante

rcas analyze --module <module> liste les Field(le=...) et validateurs Pydantic, les schémas marshmallow, les validateurs Django, les méthodes clean() et les permissions, les contrôles raise rédigés à la main, les limites nommées (MAX_QTY = 10), les comparaisons avec des nombres et les branches match. Un candidat avec drift signifie qu'un autre langage du projet (souvent le frontend) décide du même nom avec une autre valeur : une seule règles doit le décider pour les deux. Conservez la validation de type et de forme (modèles Pydantic) là où elle se trouve ; déplacez les décisions métier vers les règles. Ajoutez un test de parité qui exécute l'ancien contrôle et le moteur sur les mêmes entrées et supprimez l'ancien contrôle uniquement lorsqu'ils s'accordent.

Tests

Affirmez decision et les codes code des constats, pas le texte des messages. Les règles elles-mêmes sont testées par leurs tests golden (rcas check). En CI, exécutez rcas check et rcas compile --all avant pytest.

Référence : https://rulescascade.com/usage/python/