Metadata-Version: 2.5
Name: ddlkit-rules
Version: 0.1.0
Summary: DDL audit rules engine for ddlkit — CEL-like expression evaluator
Author-email: yangyang <15110244630@qq.com>
License: MIT
License-File: LICENSE
Keywords: audit,database,ddl,rules,sql
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Database
Classifier: Topic :: Software Development :: Quality Assurance
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: ddlkit<1,>=0.1.0
Provides-Extra: dev
Requires-Dist: mypy>=1.10; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: ruff>=0.4; extra == 'dev'
Description-Content-Type: text/markdown

# ddlkit-rules

DDL 审计规则引擎：对 `ddlkit` 解析出的 UIM 执行 CEL-like 表达式规则。

## 安装

```bash
pip install ddlkit-rules
# 或
uv add ddlkit-rules
```

依赖：`ddlkit>=0.1.0`（会自动装上）。

## 快速开始

```python
from ddlkit import parse_file
from ddlkit_rules import audit

result = parse_file("ddl.sql", dialect="dm")

violations = audit(result, [
    {
        "id": "dm.storage",
        "dialect": "dm",
        "target": "table",
        "severity": "error",
        "expr": '"STORAGE" in t.extras',
        "message": "必须指定 STORAGE 表空间",
    },
    {
        "id": "dm.pk.cluster",
        "dialect": "dm",
        "target": "constraint",
        "severity": "error",
        "expr": 'c.clustered == True',
        "message": "主键必须加 CLUSTER 关键字",
    },
    {
        "id": "naming.column.upper",
        "dialect": "*",
        "target": "column",
        "severity": "warn",
        "expr": 'col.name == col.name.upper()',
        "message": "列名建议大写",
    },
    {
        "id": "ck.order.by",
        "dialect": "ck",
        "target": "table",
        "severity": "error",
        "expr": '"ORDER BY" in t.extras',
        "message": "ClickHouse 表必须显式声明 ORDER BY",
    },
])

for v in violations:
    print(f"[{v.severity}] {v.rule_id}: {v.message} @ {v.location}")
```

## 规则定义

每条规则是一个 dict（或数据库的一行），字段如下：

| 字段 | 类型 | 说明 |
|---|---|---|
| `id` | str | 规则唯一 ID |
| `dialect` | str | 适用方言，`*` = 所有；具体值如 `dm`/`ck`/`ob`/`hive` |
| `target` | str | 校验对象：`table` / `column` / `constraint` / `index` |
| `severity` | str | 严重程度：`error` / `warn` / `info` |
| `expr` | str | CEL-like 表达式，为 `True` 表示通过 |
| `message` | str | 违规时的提示信息 |

## 表达式可用变量

### target="table" 时

| 变量 | 类型 | 说明 |
|---|---|---|
| `t.name` | str | 表名（去引号） |
| `t.name_raw` | str | 表名原文（含引号） |
| `t.quoted` | bool | 表名是否带引号 |
| `t.schema` | str | schema 名 |
| `t.catalog` | str | catalog 名 |
| `t.comment` | str\|None | 表注释 |
| `t.extras` | dict | 方言特性（键大写，如 `STORAGE`/`ORDER BY`） |
| `len(t.columns)` | int | 列数量 |

### target="column" 时

| 变量 | 类型 | 说明 |
|---|---|---|
| `col.name` | str | 列名（去引号） |
| `col.name_raw` | str | 列名原文（含引号） |
| `col.quoted` | bool | 列名是否带引号 |
| `col.type_raw` | str | 完整原始类型文本，如 `VARCHAR2(50)` |
| `col.type_name` | str | 类型名大写，如 `VARCHAR2` |
| `col.nullable` | bool\|None | 是否可空，`None` = 未声明 |
| `col.default_raw` | str\|None | 默认值原文 |
| `col.comment` | str\|None | 列注释 |
| `col.extras` | dict | 列级方言特性 |

### target="constraint" 时

| 变量 | 类型 | 说明 |
|---|---|---|
| `c.kind` | str | 约束类型：`PRIMARY KEY` / `UNIQUE` / `CHECK` / `INDEX` |
| `c.name` | str\|None | 约束名 |
| `c.clustered` | bool\|None | 是否 CLUSTER（达梦） |
| `c.columns` | list[str] | 约束涉及的列名 |

## 表达式语法

```
比较：==  !=  <  <=  >  >=
逻辑：and  or  not
成员：in  not in
字符串方法：startswith / endswith / contains / upper / lower / strip / replace / split / join
容器：len()  any()  all()  min()  max()  sum()
类型检查：isinstance(obj, "str")  isinstance(obj, "int")
```

## 从数据库加载规则

```python
import sqlite3

conn = sqlite3.connect("rules.db")
rows = conn.execute("SELECT id, dialect, target, severity, rule_expr AS expr, message FROM audit_rules").fetchall()
# rows 是 list[dict]，键名对应 Rule.from_dict 的字段

# 转成 list[dict]（字段名对齐）
rules = [
    {"id": r[0], "dialect": r[1], "target": r[2],
     "severity": r[3], "expr": r[4], "message": r[5]}
    for r in rows
]

from ddlkit_rules import audit
violations = audit(parse_result, rules)
```

## 严重程度过滤

```python
# 只看 error 级
errors = audit(result, rules, severity_filter="error")

# 只看 warn 级
warns = audit(result, rules, severity_filter="warn")
```

## License

MIT
