From cc0d80ba3a1a1798fe8463230818ed38011d2ce1 Mon Sep 17 00:00:00 2001 From: Donny Date: Thu, 5 Nov 2026 02:36:25 +0800 Subject: [PATCH] =?UTF-8?q?feat(ontology):=20=E6=9C=AC=E4=BD=93=20spec=20?= =?UTF-8?q?=E4=B8=8E=E8=B7=A8=E5=BC=95=E7=94=A8=E6=A0=A1=E9=AA=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - pydantic 模型覆盖 entities/links/actions/functions/rules/constraints - 跨引用校验：link.from/to 实体存在、rule.via 链路连得通且终点与 to 一致 - YAML 1.2 口径 loader：on/off/yes/no 不再被当成布尔（否则 on: Line 解析成 {True: Line}） - of 支持标量类型或实体名；属性支持 accepted: boolean 简写 - 19 条测试，其中 test_readme_example_loads 直接解析 README 里的 YAML - dev 组补 types-pyyaml，mypy strict 通过 --- pyproject.toml | 1 + src/adapter_python_pg/ontology/spec.py | 368 +++++++++++++++++++++++++ tests/test_ontology_spec.py | 218 +++++++++++++++ uv.lock | 11 + 4 files changed, 598 insertions(+) create mode 100644 src/adapter_python_pg/ontology/spec.py create mode 100644 tests/test_ontology_spec.py diff --git a/pyproject.toml b/pyproject.toml index cbca4db..c281dc8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,6 +25,7 @@ dev = [ "testcontainers[postgres]>=4.8", "ruff>=0.8", "mypy>=1.13", + "types-pyyaml>=6.0.12", ] [build-system] diff --git a/src/adapter_python_pg/ontology/spec.py b/src/adapter_python_pg/ontology/spec.py new file mode 100644 index 0000000..e46cc8e --- /dev/null +++ b/src/adapter_python_pg/ontology/spec.py @@ -0,0 +1,368 @@ +"""本体 spec：YAML 契约的 pydantic 模型。 + +这里的模型同时承担三件事：YAML 的语法校验、跨引用校验（link 的 from/to 是否存在、 +rule.via 是否连得通）、以及给 compiler 的类型安全输入。字段命名与 README 里的示例 +一一对应，不要在别处再定义第二套。 +""" + +from __future__ import annotations + +import re +from enum import StrEnum +from pathlib import Path +from typing import Any, Self + +import yaml +from pydantic import BaseModel, ConfigDict, Field, ValidationError, model_validator + +from adapter_python_pg.errors import OntologyError + +__all__ = [ + "ActionSpec", + "Cardinality", + "ConstraintSpec", + "EntityKind", + "EntitySpec", + "FunctionSpec", + "IdSpec", + "LinkSpec", + "Ontology", + "PropertySpec", + "PropertyType", + "RuleKind", + "RuleSpec", + "load_ontology", +] + +# 本体里的名字要能安全地进 SQL 标识符与 Cypher label +_NAME = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") + + +# YAML 1.1（PyYAML 默认）把 on/off/yes/no 当成布尔，本体里 `on: Line` 会被解析成 +# {True: 'Line'}。这里改成 1.2 口径：只有 true/false 是布尔，其余保持字符串。 +class _Yaml12SafeLoader(yaml.SafeLoader): + """只认 true/false 的安全 loader。""" + + +_Yaml12SafeLoader.yaml_implicit_resolvers = { + key: [(tag, regexp) for tag, regexp in mappings if tag != "tag:yaml.org,2002:bool"] + for key, mappings in yaml.SafeLoader.yaml_implicit_resolvers.items() +} +_Yaml12SafeLoader.add_implicit_resolver( + "tag:yaml.org,2002:bool", + re.compile(r"^(?:true|True|TRUE|false|False|FALSE)$"), + list("tTfF"), +) + + +class PropertyType(StrEnum): + STRING = "string" + NUMBER = "number" + BOOLEAN = "boolean" + DATETIME = "datetime" + ENUM = "enum" + JSON = "json" + LIST = "list" + # geo 是"全 JSONB"规则的唯一例外：compiler 为它建 geometry 生成列 + GiST 索引 + GEO = "geo" + REF = "ref" + + +class EntityKind(StrEnum): + OBJECT = "object" + TIMESERIES = "timeseries" + + +class Cardinality(StrEnum): + ONE_TO_ONE = "1:1" + ONE_TO_N = "1:N" + N_TO_ONE = "N:1" + N_TO_N = "N:N" + + +class RuleKind(StrEnum): + DERIVE_LINK = "derive_link" + DERIVE_PROPERTY = "derive_property" + + +class Effect(StrEnum): + WRITE = "write" + QUERY = "query" + EXTERNAL = "external" + + +class _Base(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True, validate_assignment=True) + + +class IdSpec(_Base): + """实体主键：调用方必须提供，adapter 不生成（幂等 upsert 与图节点 `_id` 都依赖它）。""" + + type: PropertyType = PropertyType.STRING + pattern: str | None = None + + @model_validator(mode="after") + def _only_stringish(self) -> Self: + if self.type not in {PropertyType.STRING, PropertyType.NUMBER}: + raise ValueError("id.type 只能是 string 或 number") + return self + + +class PropertySpec(_Base): + type: PropertyType + required: bool = False + unique: bool = False + index: bool = False + default: Any = None + + @model_validator(mode="before") + @classmethod + def _allow_shorthand(cls, data: object) -> object: + """允许 `accepted: boolean` 这种简写，等价于 `{type: boolean}`。""" + if isinstance(data, str): + return {"type": data} + return data + + # enum 专用 + values: tuple[str, ...] | None = None + # list 元素的类型：PropertyType 之一，或实体名（typed 结果集，action output 常用） + of: str | None = None + # 语义信息，不影响存储 + unit: str | None = None + pattern: str | None = None + # 预留：属性提升为真实列。当前不实现，只接受声明（见 AGENTS.md 拍板 3） + column: bool = False + + @model_validator(mode="after") + def _check(self) -> Self: + if self.type is PropertyType.ENUM: + if not self.values: + raise ValueError("enum 属性必须给 values") + elif self.values: + raise ValueError("只有 enum 属性可以给 values") + + if self.type is PropertyType.LIST: + if self.of is None: + raise ValueError("list 属性必须给 of（元素类型）") + elif self.of is not None: + raise ValueError("只有 list 属性可以给 of") + return self + + +class EntitySpec(_Base): + label: str | None = None + id: IdSpec = Field(default_factory=IdSpec) + properties: dict[str, PropertySpec] = Field(default_factory=dict) + kind: EntityKind = EntityKind.OBJECT + # kind=timeseries 时必填：时间列名 + 分片粒度 + time: str | None = None + partition_by: str | None = None + + @model_validator(mode="after") + def _check(self) -> Self: + for name in self.properties: + if not _NAME.match(name): + raise ValueError(f"属性名非法（需匹配 {_NAME.pattern}）：{name!r}") + if self.kind is EntityKind.TIMESERIES: + if not self.time: + raise ValueError("timeseries 实体必须声明 time") + if self.time not in self.properties: + raise ValueError(f"time={self.time!r} 未在 properties 中定义") + if self.properties[self.time].type is not PropertyType.DATETIME: + raise ValueError("time 属性类型必须是 datetime") + elif self.time is not None: + raise ValueError("只有 kind=timeseries 的实体可以声明 time") + return self + + +class LinkSpec(_Base): + type: str + from_: str = Field(alias="from") + to: str + cardinality: Cardinality = Cardinality.N_TO_ONE + properties: dict[str, PropertySpec] = Field(default_factory=dict) + + +class ActionSpec(_Base): + on: str + effect: Effect = Effect.WRITE + input: dict[str, PropertySpec] = Field(default_factory=dict) + output: dict[str, PropertySpec] = Field(default_factory=dict) + + +class FunctionSpec(_Base): + on: str + expr: str + output: PropertySpec + + +class RuleSpec(_Base): + kind: RuleKind + from_: str | None = Field(default=None, alias="from") # derive_link 用 + to: str | None = None # derive_link 用 + via: tuple[str, ...] | None = None # link 的 type 序列 + on: str | None = None # derive_property 用 + expr: str | None = None + output: PropertySpec | None = None + + @model_validator(mode="after") + def _check(self) -> Self: + if self.kind is RuleKind.DERIVE_LINK: + if not (self.from_ and self.to and self.via): + raise ValueError("derive_link 需要 from、to、via") + if self.expr or self.on or self.output: + raise ValueError("derive_link 不接受 on/expr/output") + else: + if not (self.on and self.expr and self.output): + raise ValueError("derive_property 需要 on、expr、output") + if self.from_ or self.to or self.via: + raise ValueError("derive_property 不接受 from/to/via") + return self + + +class ConstraintSpec(_Base): + on: str + expr: str + + +class Ontology(_Base): + """一份本体。跨引用校验在这里做，读进来的对象可以直接交给 compiler。""" + + ontology: str + version: int = 1 + entities: dict[str, EntitySpec] + links: dict[str, LinkSpec] = Field(default_factory=dict) + actions: dict[str, ActionSpec] = Field(default_factory=dict) + functions: dict[str, FunctionSpec] = Field(default_factory=dict) + rules: dict[str, RuleSpec] = Field(default_factory=dict) + constraints: dict[str, ConstraintSpec] = Field(default_factory=dict) + + @model_validator(mode="after") + def _check(self) -> Self: + if not self.entities: + raise ValueError("本体必须至少定义一个实体") + for name in (*self.entities, *self.links, *self.actions, *self.functions): + if not _NAME.match(name): + raise ValueError(f"名字非法（需匹配 {_NAME.pattern}）：{name!r}") + + for name, link in self.links.items(): + self._require_entity(link.from_, f"link {name}.from") + self._require_entity(link.to, f"link {name}.to") + for name, action in self.actions.items(): + self._require_entity(action.on, f"action {name}.on") + for name, func in self.functions.items(): + self._require_entity(func.on, f"function {name}.on") + for name, rule in self.rules.items(): + self._check_rule(name, rule) + for name, cons in self.constraints.items(): + self._require_entity(cons.on, f"constraint {name}.on") + self._check_of_types() + return self + + # -- 查询辅助 --------------------------------------------------------- + + def entity(self, name: str) -> EntitySpec: + try: + return self.entities[name] + except KeyError: + raise OntologyError(f"本体里没有实体 {name!r}") from None + + def timeseries_entities(self) -> dict[str, EntitySpec]: + return {k: v for k, v in self.entities.items() if v.kind is EntityKind.TIMESERIES} + + def link_types(self) -> dict[str, list[tuple[str, str]]]: + """link.type → [(from, to)]，供 rule.via 的链式校验使用。""" + edges: dict[str, list[tuple[str, str]]] = {} + for link in self.links.values(): + edges.setdefault(link.type, []).append((link.from_, link.to)) + return edges + + def _check_of_types(self) -> None: + """`of` 要么是标量类型名，要么是已定义的实体名。""" + scalar = {str(t) for t in PropertyType} + + def check(where: str, props: dict[str, PropertySpec]) -> None: + for prop_name, prop in props.items(): + if prop.of is None or prop.of in scalar: + continue + self._require_entity(prop.of, f"{where}.{prop_name}.of") + + for entity_name, entity in self.entities.items(): + check(entity_name, entity.properties) + for link_name, link in self.links.items(): + check(link_name, link.properties) + for action_name, action in self.actions.items(): + check(f"{action_name}.input", action.input) + check(f"{action_name}.output", action.output) + + # -- 内部校验 --------------------------------------------------------- + + def _require_entity(self, name: str | None, where: str) -> None: + if name is None or name not in self.entities: + raise ValueError(f"{where} 引用了不存在的实体：{name!r}") + + def _check_rule(self, name: str, rule: RuleSpec) -> None: + if rule.kind is RuleKind.DERIVE_PROPERTY: + self._require_entity(rule.on, f"rule {name}.on") + return + + self._require_entity(rule.from_, f"rule {name}.from") + self._require_entity(rule.to, f"rule {name}.to") + assert rule.from_ and rule.to and rule.via # RuleSpec 已保证 + + edges = self.link_types() + current = rule.from_ + walked: list[str] = [] + for step, link_type in enumerate(rule.via, start=1): + candidates = edges.get(link_type) + if not candidates: + raise ValueError(f"rule {name}.via 引用了不存在的 link type：{link_type!r}") + hop = next((dst for src, dst in candidates if src == current), None) + if hop is None: + starts = ", ".join(sorted({src for src, _ in candidates})) + raise ValueError( + f"rule {name}.via 链在第 {step} 步断了：{link_type!r} 没有从 " + f"{current!r} 出发的边（该 type 的起点有：{starts}）" + ) + current = hop + walked.append(link_type) + if current != rule.to: + raise ValueError( + f"rule {name} 的 via 链终点是 {current!r}，与 to={rule.to!r} 不一致" + f"（链路：{rule.from_} -{' → '.join(walked)}→ {current}）" + ) + + +def load_ontology(source: str | Path) -> Ontology: + """从 YAML 文件路径或 YAML 文本加载本体。 + + 任何结构/引用问题都收敛成 `OntologyError`，不把 pydantic 的 ValidationError + 泄漏给调用方（宿主只应 catch 一种异常）。 + """ + if isinstance(source, Path) or ("\n" not in source and source.endswith((".yml", ".yaml"))): + path = Path(source) + if not path.is_file(): + raise OntologyError(f"本体文件不存在：{path}") + text = path.read_text(encoding="utf-8") + else: + text = source + + try: + raw = yaml.load(text, Loader=_Yaml12SafeLoader) + except yaml.YAMLError as exc: + raise OntologyError(f"YAML 语法错误：{exc}") from exc + if not isinstance(raw, dict): + raise OntologyError("本体根节点必须是映射（mapping）") + + try: + return Ontology.model_validate(raw) + except ValidationError as exc: + raise OntologyError(_format_validation_error(exc)) from exc + + +def _format_validation_error(exc: ValidationError) -> str: + lines = [] + for err in exc.errors(): + loc = ".".join(str(part) for part in err["loc"]) or "" + lines.append(f"{loc}: {err['msg']}") + return "本体校验失败 —— " + "; ".join(lines) diff --git a/tests/test_ontology_spec.py b/tests/test_ontology_spec.py new file mode 100644 index 0000000..ba86ee0 --- /dev/null +++ b/tests/test_ontology_spec.py @@ -0,0 +1,218 @@ +"""本体 spec 测试。 + +其中 `test_readme_example_loads` 直接解析 README 里那段 YAML —— 文档示例一旦写坏， +这里先红，读者不会抄到坏例子。 +""" + +from __future__ import annotations + +import re +from pathlib import Path + +import pytest +import yaml + +from adapter_python_pg.errors import OntologyError +from adapter_python_pg.ontology.spec import ( + Cardinality, + EntityKind, + PropertyType, + RuleKind, + load_ontology, +) + +README = Path(__file__).resolve().parents[1] / "README.md" + + +def parse(mapping: dict[str, object]) -> object: + """走公开入口（YAML 文本 → Ontology），与真实使用路径一致。""" + return load_ontology(yaml.safe_dump(mapping, allow_unicode=True)) + + +def readme_yaml() -> str: + text = README.read_text(encoding="utf-8") + block = re.search(r"```yaml\n(.*?)```", text, re.S) + assert block, "README 里没有 yaml 代码块" + return block.group(1) + + +def minimal(**overrides: object) -> dict[str, object]: + """最简本体，供各用例覆盖局部字段。""" + base: dict[str, object] = { + "ontology": "t", + "entities": { + "A": {}, + "B": {}, + }, + "links": {"A_B": {"type": "rel", "from": "A", "to": "B"}}, + } + base.update(overrides) + return base + + +# -- README 契约 ----------------------------------------------------------- + + +def test_readme_example_loads() -> None: + onto = load_ontology(readme_yaml()) + + assert onto.ontology == "factory-twin" + assert set(onto.entities) == {"Factory", "Line", "Device", "SensorReading"} + assert set(onto.links) == {"Line_Factory", "Device_Line"} + assert onto.entities["SensorReading"].kind is EntityKind.TIMESERIES + assert onto.entities["SensorReading"].time == "ts" + assert onto.entities["Factory"].properties["location"].type is PropertyType.GEO + assert onto.entities["Line"].properties["status"].values == ("running", "idle", "fault") + assert onto.links["Device_Line"].cardinality is Cardinality.N_TO_ONE + assert onto.links["Device_Line"].from_ == "Device" + assert onto.rules["Device_inFactory"].kind is RuleKind.DERIVE_LINK + assert onto.rules["Device_inFactory"].via == ("mounted_on", "belongs_to") + + +def test_readme_rule_chain_resolves_device_to_factory() -> None: + onto = load_ontology(readme_yaml()) + assert onto.link_types()["mounted_on"] == [("Device", "Line")] + assert onto.link_types()["belongs_to"] == [("Line", "Factory")] + + +def test_timeseries_entities_helper() -> None: + onto = load_ontology(readme_yaml()) + assert list(onto.timeseries_entities()) == ["SensorReading"] + + +# -- 跨引用校验 ------------------------------------------------------------ + + +def test_link_target_must_exist() -> None: + bad = minimal(links={"A_B": {"type": "rel", "from": "A", "to": "Nope"}}) + with pytest.raises(OntologyError, match=r"link A_B\.to 引用了不存在的实体"): + parse(bad) + + +def test_link_source_must_exist() -> None: + bad = minimal(links={"A_B": {"type": "rel", "from": "Nope", "to": "B"}}) + with pytest.raises(OntologyError, match=r"link A_B\.from 引用了不存在的实体"): + parse(bad) + + +def test_rule_via_must_be_connected() -> None: + bad = minimal( + rules={ + "backwards": { + "kind": "derive_link", + "from": "B", + "to": "A", + "via": ["rel"], # rel 只从 A 出发，B 走不动 + } + } + ) + with pytest.raises(OntologyError, match="via 链在第 1 步断了"): + parse(bad) + + +def test_rule_chain_end_must_match_to() -> None: + bad = minimal( + entities={"A": {}, "B": {}, "C": {}}, + links={ + "A_B": {"type": "rel", "from": "A", "to": "B"}, + "B_C": {"type": "rel2", "from": "B", "to": "C"}, + }, + rules={ + "short": {"kind": "derive_link", "from": "A", "to": "C", "via": ["rel"]}, + }, + ) + with pytest.raises(OntologyError, match="via 链终点是 'B'"): + parse(bad) + + +def test_rule_via_unknown_type() -> None: + bad = minimal( + rules={"r": {"kind": "derive_link", "from": "A", "to": "B", "via": ["nope"]}}, + ) + with pytest.raises(OntologyError, match="不存在的 link type"): + parse(bad) + + +# -- 字段级校验 ------------------------------------------------------------ + + +def test_enum_requires_values() -> None: + bad = minimal(entities={"A": {"properties": {"s": {"type": "enum"}}}}) + with pytest.raises(OntologyError, match="enum 属性必须给 values"): + parse(bad) + + +def test_values_only_on_enum() -> None: + bad = minimal(entities={"A": {"properties": {"s": {"type": "string", "values": ["x"]}}}}) + with pytest.raises(OntologyError, match="只有 enum 属性可以给 values"): + parse(bad) + + +def test_list_requires_of() -> None: + bad = minimal(entities={"A": {"properties": {"tags": {"type": "list"}}}}) + with pytest.raises(OntologyError, match="list 属性必须给 of"): + parse(bad) + + +def test_timeseries_requires_declared_time() -> None: + bad = minimal( + entities={ + "A": {"kind": "timeseries", "time": "ts"}, # properties 里没有 ts + } + ) + with pytest.raises(OntologyError, match="未在 properties 中定义"): + parse(bad) + + +def test_timeseries_time_must_be_datetime() -> None: + bad = minimal( + entities={ + "A": { + "kind": "timeseries", + "time": "ts", + "properties": {"ts": {"type": "string"}}, + } + } + ) + with pytest.raises(OntologyError, match="time 属性类型必须是 datetime"): + parse(bad) + + +def test_unknown_field_is_rejected() -> None: + """拼错字段名必须报错，而不是被静默忽略。""" + bad = minimal(entities={"A": {"properties": {"x": {"type": "string", "indx": True}}}}) + with pytest.raises(OntologyError, match="indx"): + parse(bad) + + +def test_bad_identifier() -> None: + bad = minimal(entities={"A; drop table x": {}}) + with pytest.raises(OntologyError, match="名字非法"): + parse(bad) + + +# -- 加载入口 -------------------------------------------------------------- + + +def test_load_from_text_and_path(tmp_path: Path) -> None: + text = readme_yaml() + assert load_ontology(text).ontology == "factory-twin" + + path = tmp_path / "onto.yaml" + path.write_text(text, encoding="utf-8") + assert load_ontology(path).ontology == "factory-twin" + + +def test_missing_file(tmp_path: Path) -> None: + with pytest.raises(OntologyError, match="本体文件不存在"): + load_ontology(tmp_path / "nope.yaml") + + +def test_yaml_syntax_error() -> None: + with pytest.raises(OntologyError, match="YAML 语法错误"): + load_ontology("ontology: x\nentities: [unclosed") + + +def test_root_must_be_mapping() -> None: + with pytest.raises(OntologyError, match="根节点必须是映射"): + load_ontology("- a\n- b\n") diff --git a/uv.lock b/uv.lock index c966265..bd2fcba 100644 --- a/uv.lock +++ b/uv.lock @@ -31,6 +31,7 @@ dev = [ { name = "pytest-asyncio" }, { name = "ruff" }, { name = "testcontainers" }, + { name = "types-pyyaml" }, ] [package.metadata] @@ -51,6 +52,7 @@ dev = [ { name = "pytest-asyncio", specifier = ">=0.24" }, { name = "ruff", specifier = ">=0.8" }, { name = "testcontainers", extras = ["postgres"], specifier = ">=4.8" }, + { name = "types-pyyaml", specifier = ">=6.0.12" }, ] [[package]] @@ -1225,6 +1227,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/00/7e/424aac8b355597835deb333e757a0e94b5ccf38ad00f07fe6ed1f4e17c88/testcontainers-4.15.0-py3-none-any.whl", hash = "sha256:8796c14e76604031ad39cf0ed3b8e9806283a1fbf5270965c2b1c594caa31b74", size = 160771, upload-time = "2026-07-24T23:08:00.13Z" }, ] +[[package]] +name = "types-pyyaml" +version = "6.0.12.20261103" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/d9/82/2ed4d215a2840b467a9a864f03b0d21904250e28f5e1f006ad36fba0311a/types_pyyaml-6.0.12.20261103.tar.gz", hash = "sha256:7a1b5f837881dd35d5448b5f18275164f89f206185e31e8eae6e0b42607bc001", size = 18911, upload-time = "2026-11-03T08:45:23.714Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/e8/10/4353eb86278f6a59aeb9fe2be4d979c20b110b4009fa455a3ea6467cf5cb/types_pyyaml-6.0.12.20261103-py3-none-any.whl", hash = "sha256:444654c61f635dab82c092a511aa96a0ce31164810784b9a37314e3e56532ac8", size = 21299, upload-time = "2026-11-03T08:45:22.749Z" }, +] + [[package]] name = "typing-extensions" version = "4.16.0"