Python에서의 Rule Cascade
패키지: pip install rule-cascade (Python 3.11+; rule_cascade 임포트). 번들은 rcas compile --all 명령으로 build/rules/에 생성됩니다. 런타임에 YAML을 평가하지 말고, 코드에 규칙을 복사하지 마세요.
한 번 로드, 요청마다 평가
import json
from rule_cascade import RuleSet, RequestError
# 시작 시 (모듈 레벨, FastAPI lifespan, Django AppConfig.ready). 잘못된 번들은 서비스 중단을 유발합니다.
with open("build/rules/<ruleset-id>.bundle.json", encoding="utf-8") as f:
rules = RuleSet.from_bundle(json.load(f))
데코레이터 사용을 권장합니다 (1.0.0-alpha.7부터): 요청을 구성하고, 함수 실행 전에 규칙을 적용하며, RuleViolation을 발생시킵니다. problem()은 아래의 422 응답을 반환합니다.
from rule_cascade import enforce_rules, current_evaluation
from rule_cascade.fastapi import install # 또는 rule_cascade.flask.init_app(app), 또는 Django 미들웨어
install(app) # RuleViolation -> 422, RequestError -> 400
@app.post("/orders", status_code=201)
@enforce_rules(rules, entity="Order", operation="create", data="order", actor="user") # 라우트 데코레이터 아래에 위치
async def create_order(order: Order, user: User = Depends(current_user)):
evaluation = current_evaluation() # 허용된 결과: 경고, 계산된 값, 명령
rules.enforce(request, "server")는 데코레이터 없이 동일한 작업을 수행합니다. 내부적으로 rules.evaluate(request, "server")는 decision (allow 또는 deny), findings, effects, commands를 포함하는 딕셔너리를 반환합니다. 잘못된 요청에는 RequestError를 발생시킵니다(400 응답). 모든 상태 변경 엔드포인트는 다음 네 단계를 따릅니다:
- 서버에서 평가
{"entity", "operation", "data", "original", "actor", "resolutions"}를 사용;actor는 인증에서 가져오며({"id", "roles"}), 호출자가 보낸 필드에서 직접 가져오지 않습니다. - 거부 시 거절 422 문제 세부 정보(
application/problem+json,type: "urn:rule-cascade:rule-violation")와 함께; 결과를 포함하여 클라이언트가findings[].fields를 입력에 매핑할 수 있도록 합니다. - 저장
compute효과(effect["type"] == "value")를 적용한 후: 서버의 값이 저장됩니다. - 커밋 후
result["commands"]실행idempotencyKey당 한 번씩 (아웃박스 테이블 또는 중복 제거가 있는 작업 큐 사용).
프레임워크 배치: FastAPI - 의존성(Depends(get_rules))과 RuleViolation을 422로 변환하는 예외 핸들러; Django - 각 뷰나 직렬화기 validate()에서 호출되는 서비스 함수와 422 처리를 위한 미들웨어 또는 DRF 예외 핸들러; Flask - 앱 팩토리에서 생성된 확장 객체. Celery 및 배치 작업은 모든 쓰기 전에 동일한 방식으로 평가합니다.
RuleSetHolder(load_bundle, interval=300) (또는 cron="0 * * * *")는 백그라운드에서 번들을 다시 로드하고, 다시 로드 실패 시 마지막으로 잘 작동한 규칙을 유지합니다.
많은 레코드
파일 및 큐의 경우 rcas engine을 한 번 시작하고 JSON Lines 요청을 스트리밍하거나, rules.evaluate를 루프에서 호출합니다: 두 경우 모두 번들을 한 번 로드합니다. 자세한 내용은 https://rulescascade.com/examples/batch/를 참조하세요.
기존 검증 대체
rcas analyze --module <module> 명령은 Pydantic Field(le=...) 및 validators, marshmallow 스키마, Django validators, clean() 메서드 및 권한, 수동 raise 검사, 명명된 제한(MAX_QTY = 10), 숫자 비교 및 match 분기를 나열합니다. drift가 있는 후보는 프로젝트의 다른 언어(주로 프론트엔드)가 동일한 이름을 다른 값으로 결정함을 의미합니다: 하나의 규칙 세트가 두 경우 모두를 결정해야 합니다. 유형 및 형상 검증(Pydantic 모델)은 그대로 두고, 비즈니스 의사 결정을 규칙으로 이동합니다. 이전 검사와 엔진을 동일한 입력으로 실행하는 패리티 테스트를 추가하고, 일치하는 경우에만 이전 검사를 삭제합니다.
테스트
decision 및 finding code를 검증하고, 메시지 텍스트는 검증하지 않습니다. 규칙 자체는 golden 테스트(rcas check)로 테스트됩니다. CI에서는 pytest 전에 rcas check와 rcas compile --all을 실행합니다.