mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 00:11:59 +00:00
规则 DSL 编译成只读 SQL，并用 fixture 验证
This commit is contained in:
ThearchyHelios
@@ -286,6 +286,27 @@ async def caliber_descriptor(workflow_id: str, version: int, node_id: str,
|
||||
return made
|
||||
|
||||
|
||||
class RuleCompileIn(BaseModel):
|
||||
"""规则 DSL 编译：表 + 结构化条件 → 只读 SQL；fixtures 可选，用内存 SQLite 跑一遍。"""
|
||||
|
||||
rule: dict[str, Any]
|
||||
table: str
|
||||
fixtures: list[dict[str, Any]] = Field(default_factory=list)
|
||||
|
||||
|
||||
@router.post("/rules/compile")
|
||||
async def compile_rule_endpoint(payload: RuleCompileIn) -> dict[str, Any]:
|
||||
"""把结构化规则编译成 SQL，并（给了 fixtures 时）用样例行验证。只生成 SELECT，不碰业务库。"""
|
||||
from app.data.rule_compile import RuleInvalid, compile_rule, run_fixtures
|
||||
|
||||
try:
|
||||
sql = compile_rule(payload.rule, payload.table)
|
||||
checks = run_fixtures(payload.rule, payload.table, payload.fixtures)
|
||||
except RuleInvalid as e:
|
||||
raise HTTPException(422, str(e)) from None
|
||||
return {"sql": sql, "fixtures": checks}
|
||||
|
||||
|
||||
class TrialIn(BaseModel):
|
||||
"""试跑只收入口字段的值；没写的按口径里的缺省值补。"""
|
||||
|
||||
|
||||
@@ -0,0 +1,130 @@
|
||||
"""规则 DSL 的编译与 fixture 校验：结构化条件 → 一条只读 SQL，发布前用样例行跑一遍。
|
||||
|
||||
和口径卡里的 `kind=rule` 不是同一层：口径卡里的规则在单元格上确定性求值（不需要查库）；这里给作者和发布检查用，
|
||||
把「表 + 条件 → then / else」编译成 SQL，再用 fixture 验证条件写对了。编译只生成 SELECT，不执行任何写操作。
|
||||
|
||||
规则形状：
|
||||
{
|
||||
"inputs": {"status": "status", "amount": "amount"}, # 名字 → 表里的列
|
||||
"all": [{"input": "status", "op": "eq", "value": "paid"},
|
||||
{"input": "amount", "op": "gte", "value": 50}],
|
||||
"then": 1, "else": 0
|
||||
}
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import sqlite3
|
||||
from typing import Any, Iterable, Mapping
|
||||
|
||||
OPS = ("eq", "ne", "gt", "gte", "lt", "lte", "in", "not_in")
|
||||
_IDENT = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
|
||||
|
||||
|
||||
class RuleInvalid(ValueError):
|
||||
"""规则本身写错了：message 给人看。"""
|
||||
|
||||
|
||||
def _ident(value: Any, what: str) -> str:
|
||||
text = str(value or "").strip()
|
||||
if not _IDENT.match(text):
|
||||
raise RuleInvalid(f"{what}「{value}」不是合法的标识符（字母或下划线开头，只含字母、数字、下划线）")
|
||||
return text
|
||||
|
||||
|
||||
def _literal(exp: Any, value: Any) -> Any:
|
||||
if isinstance(value, bool):
|
||||
return exp.Boolean(this=value)
|
||||
if isinstance(value, (int, float)):
|
||||
return exp.Literal.number(value)
|
||||
if value is None:
|
||||
return exp.Null()
|
||||
return exp.Literal.string(str(value))
|
||||
|
||||
|
||||
def _condition(exp: Any, cond: Mapping[str, Any], inputs: Mapping[str, str]) -> Any:
|
||||
name = str(cond.get("input") or "")
|
||||
if name not in inputs:
|
||||
raise RuleInvalid(f"条件指向了没声明的输入「{name}」")
|
||||
column = exp.column(_ident(inputs[name], "列名"))
|
||||
op = str(cond.get("op") or "")
|
||||
if op not in OPS:
|
||||
raise RuleInvalid(f"条件运算符只能是 {'、'.join(OPS)}，当前为「{op}」")
|
||||
if op in ("in", "not_in"):
|
||||
values = cond.get("values")
|
||||
if not isinstance(values, list) or not values:
|
||||
raise RuleInvalid("in / not_in 需要非空的 values 列表")
|
||||
node = exp.In(this=column, expressions=[_literal(exp, v) for v in values])
|
||||
return node if op == "in" else exp.Not(this=node)
|
||||
if "value" not in cond:
|
||||
raise RuleInvalid("条件需要 value")
|
||||
right = _literal(exp, cond["value"])
|
||||
return {
|
||||
"eq": exp.EQ, "ne": exp.NEQ, "gt": exp.GT, "gte": exp.GTE, "lt": exp.LT, "lte": exp.LTE,
|
||||
}[op](this=column, expression=right)
|
||||
|
||||
|
||||
def compile_rule(rule: Mapping[str, Any], table: str, *, dialect: str | None = None) -> str:
|
||||
"""结构化规则 → `SELECT CASE WHEN … THEN … ELSE … END AS value FROM <table>`。"""
|
||||
try:
|
||||
import sqlglot
|
||||
from sqlglot import exp
|
||||
except ImportError as e: # pragma: no cover - 依赖缺失时给一句人话
|
||||
raise RuleInvalid("没有安装 sqlglot，无法编译规则") from e
|
||||
table_name = _ident(table, "表名")
|
||||
if not isinstance(rule, Mapping):
|
||||
raise RuleInvalid("规则需要是一个对象")
|
||||
inputs = rule.get("inputs")
|
||||
if not isinstance(inputs, Mapping) or not inputs:
|
||||
raise RuleInvalid("规则需要 inputs：名字 → 表里的列")
|
||||
columns = {str(k): _ident(v, "列名") for k, v in inputs.items()}
|
||||
conditions = rule.get("all")
|
||||
if not isinstance(conditions, list) or not conditions:
|
||||
raise RuleInvalid("规则需要 all：至少一条条件")
|
||||
where = None
|
||||
for cond in conditions:
|
||||
node = _condition(exp, cond, columns)
|
||||
where = node if where is None else exp.And(this=where, expression=node)
|
||||
case = exp.Case(
|
||||
ifs=[exp.If(this=where, true=_literal(exp, rule.get("then", 1)))],
|
||||
default=_literal(exp, rule.get("else", 0)),
|
||||
)
|
||||
query = exp.select(exp.alias_(case, "value")).from_(exp.table_(table_name))
|
||||
return query.sql(dialect=dialect or "sqlite")
|
||||
|
||||
|
||||
def run_fixtures(rule: Mapping[str, Any], table: str, fixtures: Iterable[Mapping[str, Any]], *,
|
||||
dialect: str = "sqlite") -> list[dict[str, Any]]:
|
||||
"""用一条只读 SQL 在内存 SQLite 上跑 fixture：每项 {rows, expected}，返回 [{ok, expected, actual}]。
|
||||
|
||||
fixture 的 rows 是列名 → 值；表结构按第一行推（整数 / 实数 / 文本）。空 fixture 返回空列表。
|
||||
"""
|
||||
items = [dict(f) for f in fixtures]
|
||||
if not items:
|
||||
return []
|
||||
sql = compile_rule(rule, table, dialect="sqlite" if dialect != "sqlite" else dialect)
|
||||
db = sqlite3.connect(":memory:")
|
||||
try:
|
||||
columns: dict[str, str] = {}
|
||||
for item in items:
|
||||
for row in item.get("rows") or []:
|
||||
for name, value in row.items():
|
||||
columns.setdefault(str(name), "INTEGER" if isinstance(value, int) and not isinstance(value, bool)
|
||||
else "REAL" if isinstance(value, float) else "TEXT")
|
||||
if not columns:
|
||||
raise RuleInvalid("fixture 需要至少一行 rows")
|
||||
db.execute(f"CREATE TABLE {table} (" + ", ".join(f"{name} {kind}" for name, kind in columns.items()) + ")")
|
||||
out: list[dict[str, Any]] = []
|
||||
for item in items:
|
||||
db.execute(f"DELETE FROM {table}")
|
||||
for row in item.get("rows") or []:
|
||||
db.execute(f"INSERT INTO {table} ({', '.join(row)}) VALUES ({', '.join('?' for _ in row)})",
|
||||
list(row.values()))
|
||||
actual = db.execute(sql).fetchone()[0]
|
||||
out.append({"ok": actual == item.get("expected"), "expected": item.get("expected"), "actual": actual})
|
||||
return out
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
__all__ = ["OPS", "RuleInvalid", "compile_rule", "run_fixtures"]
|
||||
@@ -0,0 +1,58 @@
|
||||
"""规则 DSL 编译（app/data/rule_compile.py、POST /api/ledger/rules/compile）：结构化条件 → 只读 SQL，fixture 验证。"""
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from httpx import ASGITransport, AsyncClient
|
||||
|
||||
from app.data.rule_compile import RuleInvalid, compile_rule, run_fixtures
|
||||
from app.main import app
|
||||
|
||||
RULE = {"inputs": {"status": "status", "amount": "amount"},
|
||||
"all": [{"input": "status", "op": "eq", "value": "paid"},
|
||||
{"input": "amount", "op": "gte", "value": 50}],
|
||||
"then": 1, "else": 0}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def client():
|
||||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
|
||||
yield c
|
||||
|
||||
|
||||
def test_compile_rule_is_a_select_with_a_case():
|
||||
sql = compile_rule(RULE, "orders")
|
||||
assert sql.lower().startswith("select") and "case when" in sql.lower()
|
||||
assert "from orders" in sql.lower()
|
||||
assert "insert" not in sql.lower() and "update" not in sql.lower() and "delete" not in sql.lower()
|
||||
|
||||
|
||||
def test_fixtures_run_against_the_compiled_sql():
|
||||
checks = run_fixtures(RULE, "orders", [
|
||||
{"rows": [{"status": "paid", "amount": 100}], "expected": 1},
|
||||
{"rows": [{"status": "paid", "amount": 10}], "expected": 0},
|
||||
{"rows": [{"status": "open", "amount": 100}], "expected": 0},
|
||||
])
|
||||
assert [c["ok"] for c in checks] == [True, True, True]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("rule", [
|
||||
{"inputs": {}, "all": [{"input": "x", "op": "eq", "value": 1}]},
|
||||
{"inputs": {"x": "x"}, "all": [{"input": "x", "op": "like", "value": 1}]},
|
||||
{"inputs": {"x": "x"}, "all": [{"input": "y", "op": "eq", "value": 1}]},
|
||||
])
|
||||
def test_bad_rules_are_refused(rule):
|
||||
with pytest.raises(RuleInvalid):
|
||||
compile_rule(rule, "orders")
|
||||
|
||||
|
||||
async def test_the_compile_endpoint_returns_sql_and_fixture_results(client):
|
||||
res = await client.post("/api/ledger/rules/compile", json={
|
||||
"rule": RULE, "table": "orders",
|
||||
"fixtures": [{"rows": [{"status": "paid", "amount": 100}], "expected": 1},
|
||||
{"rows": [{"status": "paid", "amount": 10}], "expected": 0}],
|
||||
})
|
||||
assert res.status_code == 200, res.text
|
||||
body = res.json()
|
||||
assert body["sql"].lower().startswith("select") and [c["ok"] for c in body["fixtures"]] == [True, True]
|
||||
bad = await client.post("/api/ledger/rules/compile", json={"rule": {"inputs": {}}, "table": "orders"})
|
||||
assert bad.status_code == 422
|
||||
Reference in New Issue
Block a user