Rule Cascade em Python
Pacote: pip install rule-cascade (Python 3.11+; import rule_cascade). Os pacotes vêm de
rcas compile --all em build/rules/. Nunca avalie o YAML em tempo de execução e nunca copie uma regra para o código.
Carregue uma vez, avalie por requisição
import json
from rule_cascade import RuleSet, RequestError
# Na inicialização (nível de módulo, lifespan do FastAPI, AppConfig.ready do Django). Um pacote ruim levanta erro: não sirva.
with open("build/rules/<ruleset-id>.bundle.json", encoding="utf-8") as f:
rules = RuleSet.from_bundle(json.load(f))
Prefira o decorador (desde 1.0.0-alpha.7): ele constrói a requisição, aplica as regras antes da função
ser executada e levanta RuleViolation, cujo problem() é a resposta 422 abaixo.
from rule_cascade import enforce_rules, current_evaluation
from rule_cascade.fastapi import install # ou rule_cascade.flask.init_app(app), ou o middleware do Django
install(app) # RuleViolation -> 422, RequestError -> 400
@app.post("/orders", status_code=201)
@enforce_rules(rules, entity="Order", operation="create", data="order", actor="user") # abaixo do decorador de rota
def create_order(order: Order, user: User = Depends(current_user)):
evaluation = current_evaluation() # o resultado permitido: avisos, valores computados, comandos
rules.enforce(request, "server") faz o mesmo sem um decorador. Por baixo dos panos,
rules.evaluate(request, "server") retorna um dicionário: decision (allow ou deny), findings,
effects, commands. Ele levanta RequestError para uma requisição malformada (resposta 400). Todo
endpoint que altera estado segue quatro passos:
- Avalie no servidor com
{"entity", "operation", "data", "original", "actor", "resolutions"};actorvem da autenticação ({"id", "roles"}), nunca de um campo enviado pelo chamador. - Recuse em caso de negação com 422 detalhes do problema (
application/problem+json,type: "urn:rule-cascade:rule-violation") que incluem o resultado, para que clientes possam mapearfindings[].fieldspara as entradas. - Persista, após aplicar efeitos
compute(effect["type"] == "value"): o valor do servidor é armazenado. - Execute
result["commands"]após o commit, uma vez poridempotencyKey(uma tabela de outbox, ou uma fila de tarefas com deduplicação).
Posicionamento nos frameworks: FastAPI como dependência (Depends(get_rules)) e um manipulador de exceções que transforma
RuleViolation no 422; Django como função de serviço chamada por toda view ou validate() do serializer
da entidade, mais middleware ou manipulador de exceções DRF para o 422; Flask como objeto de extensão
criado na factory da aplicação. Celery e jobs em lote avaliam antes de cada escrita da mesma forma.
RuleSetHolder(load_bundle, interval=300) (ou cron="0 * * * *") recarrega pacotes em segundo plano
e mantém as últimas regras boas quando uma recarga falha.
Muitos registros
Para arquivos e filas, inicie rcas engine uma vez e transmita requisições JSON Lines através dele, ou chame
rules.evaluate em um loop: carregue o pacote uma vez de qualquer forma. Veja
https://rulescascade.com/examples/batch/.
Substituindo validações existentes
rcas analyze --module <module> lista Field(le=...) e validadores do Pydantic, esquemas marshmallow,
validadores do Django, métodos clean() e permissões, verificações manuais com raise, limites
nomeados (MAX_QTY = 10), comparações com números e ramos match. Um candidato com drift significa que outra linguagem do projeto (geralmente o front end) decide o mesmo nome com outro valor: um único conjunto de regras deve decidir por ambos. Mantenha a validação de tipo e forma (modelos Pydantic) onde está; mova decisões de negócio para as regras. Adicione um teste de paridade que executa a verificação antiga e o motor nas mesmas entradas e
exclua a verificação antiga apenas quando concordarem.
Testes
Afirme decision e códigos dos findings, não textos de mensagens. As próprias regras são testadas por seus
testes dourados (rcas check). No CI execute rcas check e rcas compile --all antes de pytest.
Referência: https://rulescascade.com/usage/python/