Files
adapter_python_pg/README.md
T
Donny ace505cc6b chore: 搭建项目骨架
- pyproject:hatchling + src 布局,核心依赖 psycopg/pydantic/pyyaml/cel-python
- FastAPI 归入 [api] 可选依赖,开发依赖走 PEP 735 dependency-groups
- 建立分层目录:ontology/compiler/storage/repository/graph/timeseries/expr/rules/api
- errors.py 统一异常基类;py.typed 标记类型完整
- ruff(100) + pytest 配置;冒烟测试校验核心包不引入 fastapi/uvicorn
2026-11-05 02:18:10 +08:00

13 KiB
Raw Blame History

adapter_python_pg

基于 ontology(本体)数据基座的中间层适配器:用 YAML 定义世界,用一套接口读写它。

面向其他项目的底层对接件,不是业务系统,也不是通用 ORM。上层项目只需要一份本体 YAML + 本适配器,就能对实体、关系、属性做对象级操作,而不用关心底下是文档表、图还是时序表。

状态:设计稿(v0.1)。仓库当前只有初始提交,实现尚未开始。本文是接口与存储映射的契约,先定契约再写代码。

工程约定(技术栈、硬规则、目录、命令、测试策略)见 AGENTS.md —— 那是给编码工具和协作者看的执行约束,本文只讲"是什么、为什么"。


1. 解决的问题

常规做法里,"数据"和"数据的意思"是两套东西:SQL 建表定义结构,业务代码里再写一遍实体、关系、校验。两套东西会漂移,且跨项目无法复用。

这里把语义提到最前面:本体 YAML 是唯一事实来源,适配器负责把这份声明翻译成存储结构和查询。上层拿到的不是 cursor,是实体对象。

  上层项目(业务服务 / 分析服务 / 数字孪生前端 / 其他 adapter)
                    │  实体 / 对象 / 语义接口
                    ▼
   ┌───────────────────────────────────────────┐
   │  adapter_python_pg                        │
   │  ├─ Ontology Loader     解析 YAML 契约     │
   │  ├─ Schema Compiler     映射到存储与索引   │
   │  ├─ Repository          实体 CRUD / 批量   │
   │  ├─ Query Planner       SQL / Cypher / 混合│
   │  ├─ Rule Engine         规则派生与约束校验 │
   │  └─ Action Runtime      Actions 编排      │
   └───────────────────────────────────────────┘
                    │  SQL / openCypher / 时序写入
                    ▼
   PostgreSQL 18 + TimescaleDB + Apache AGE
   (文档 JSONB · 图 · hypertable 三合一同库)

2. 设计原则

原则 含义
YAML 即契约 实体、关系、属性、行为全部声明在 YAML 里;改本体只改 YAML
只加载 YAML 就能用 不做代码生成、不要求手写 DDL;运行期解释本体,按需建存储结构
对象而非行 对外只有实体对象与语义方法,SQL 与 Cypher 是内部实现细节
可嵌入 既能作为库被其他 Python 项目 import,也能作为 FastAPI 服务独立部署
不持有业务 不放业务规则代码;业务规则也写在 YAML 里(Rules / Constraints)
同库三模 文档、图、时序落在同一个 PostgreSQL 实例,事务与 JOIN 天然一致

3. 本体 YAML

一份本体描述"世界里有什么、怎么连、能做什么"。以下是一个数字孪生场景的最小片段:

ontology: factory-twin
version: 1

entities:
  Factory:
    label: 工厂
    id: { type: string, pattern: "^F-[0-9]{4}$" }
    properties:
      name:      { type: string, required: true }
      location:  { type: geo, index: true }
      capacity:  { type: number, unit: unit/day }

  Line:
    label: 产线
    properties:
      name:     { type: string, required: true }
      status:   { type: enum, values: [running, idle, fault], default: idle }

  Device:
    label: 设备
    properties:
      serial:   { type: string, unique: true }
      model:    { type: string }
      installed_at: { type: datetime }
      meta:     { type: json }          # 不固定的扩展字段

  SensorReading:                         # 时序实体
    kind: timeseries
    label: 传感器读数
    time: ts
    partition_by: 7 days
    properties:
      ts:     { type: datetime, required: true }
      metric: { type: string, required: true }
      value:  { type: number, required: true }
      tags:   { type: json, index: true }

links:
  Line_Factory:
    type: belongs_to
    from: Line
    to: Factory
    cardinality: N:1
  Device_Line:
    type: mounted_on
    from: Device
    to: Line
    cardinality: N:1
    properties:
      since: { type: datetime }

actions:
  startLine:
    on: Line
    input:  { deviceIds: { type: list, of: string } }
    output: { accepted: boolean }
    effect: write                        # write | query | external
  queryDevicesByFactory:
    on: Factory
    output: { devices: { type: list, of: Device } }
    effect: query

functions:                               # 纯函数,不落库
  utilization:
    on: Line
    expr: "runningHours / capacityHours"
    output: { type: number, unit: percent }

rules:                                   # 推理:派生关系或属性
  Device_inFactory:
    kind: derive_link
    from: Device
    to: Factory
    via: [mounted_on, belongs_to]        # 传递闭包
  Line_atRisk:
    kind: derive_property
    on: Line
    expr: "faultCount(1h) > 3 and status == 'running'"
    output: { type: boolean }

constraints:
  SensorReading_positive:
    on: SensorReading
    expr: "value >= 0"
  Line_single_factory:
    on: Line
    expr: "count(belongs_to) == 1"

字段语义:

元素 作用 关键字段
entities 定义实体与属性 id、properties.type、required、unique、index、kind
links 定义关系 from/to、type、cardinality、边属性
actions 可执行动作 on、input/output、effect
functions 计算属性 expr(受限表达式,不执行任意代码)
rules 推理 derive_link(关系传递)、derive_property(派生属性)
constraints 写入校验 expr,写前校验、失败即拒绝

属性类型:string number boolean datetime enum json geo ref。type: json 是留给本体也描述不了的字段的逃生口。

4. 存储映射

加载 YAML 时由 Schema Compiler 推导存储,无需人工建表:

本体元素 落到哪 引擎
entities 的属性集合 docs(collection, doc JSONB) 一行一实体 JSONB + GIN
links ag_catalog 图节点与边 Apache AGE
kind: timeseries 实体 hypertable,按 partition_by 分片 TimescaleDB
index: true / unique GIN(jsonb_path_ops) / 唯一约束 原生索引
rules.derive_link 查询时展开,或物化到图里(可配) AGE / SQL
constraints 写前表达式校验 SQL 表达式

一个实体同时存在两种表示是刻意的:文档视图适合整实体读写与属性过滤,图视图适合多跳遍历。两者用同一 id 对齐,写入在同一事务里完成。

5. 写入

两种写入路径,对应两种数据形状:

文档写入——实体是整体,属性不固定:

repo.save(Device(id="D-0001", serial="SN-88", model="X200", meta={"vendor": "acme"}))
repo.save_many(devices)
repo.patch("D-0001", {"meta.vendor": "acme-new"})  # 部分更新

时序写入——数字孪生的遥测流,按时间追加:

ts = adapter.timeseries("SensorReading")
ts.write([{"ts": now, "metric": "temp", "value": 36.5, "tags": {"device": "D-0001"}}])
ts.write_batch(stream, batch_size=1000)  # 批量/流式

写入语义约定:save 按 id 幂等 upsert;link() 幂等建边;constraints 在提交前校验,失败抛 ConstraintViolation;批量写入单事务,超限自动分块。

6. 查询

三种查询,同一套实体对象返回:

属性查询(SQL 侧)

adapter.find(Device, model="X200", installed_after="2025-01-01")
adapter.find(Device, where={"meta.vendor": "acme"}, limit=50)
adapter.aggregate("SensorReading", metric="temp", bucket="1m", agg="avg")

图查询(Cypher 侧)

adapter.traverse(Device, link="mounted_on", to=Line, hops="1..3")
adapter.path(Device("D-0001"), Factory, max_hops=4)

混合与推理

# 规则派生的关系直接可用,不需要先物化
adapter.find(Device, rule="Device_inFactory", factory="F-0001")
# 属性来自 function,与库内字段一起返回
adapter.find(Line, select=["status", "utilization"])

推理的边界先划清楚:规则驱动的确定性推理——关系传递闭包(via)、派生属性(受限表达式)、约束推导;不做概率推理、不做模型推理。跑在图查询或 SQL 里,结果可复现。

7. 对外接口

语义层 API,不暴露 SQL / Cypher:

from adapter_python_pg import Adapter, ConstraintViolation

adapter = Adapter.load("ontology/factory-twin.yaml", dsn="postgresql://...")

repo = adapter.repository("Device")  # 实体仓储
repo.save(obj)
repo.get(id)
repo.find(**where)
repo.patch(id, patch)
repo.delete(id)

graph = adapter.graph()  # 图操作
graph.link(src, "mounted_on", dst, since=...)
graph.traverse(...)
graph.path(...)

ts = adapter.timeseries("SensorReading")  # 时序
ts.write(rows)
ts.query(bucket="5m", agg="avg", **filters)

adapter.invoke("startLine", on="L-01", input={"deviceIds": ["D-0001"]})  # Actions
adapter.functions("Line")["utilization"]  # Functions
adapter.validate("SensorReading", row)  # Constraints

FastAPI 侧不提供独立服务:build_router() 返回一个 APIRouter,由宿主项目的 app.include_router() 挂进去。路由从本体自动生成,同一份 YAML 不重复声明:

from fastapi import FastAPI
from adapter_python_pg import Adapter, build_router

adapter = Adapter.load("ontology/factory-twin.yaml", dsn=settings.pg_dsn)

app = FastAPI()
app.include_router(build_router(adapter), prefix="/ontology")
GET    /ontology/entities/{entity}                列表 / 过滤
GET    /ontology/entities/{entity}/{id}           单实体(可选 ?expand=links)
POST   /ontology/entities/{entity}                写入
PATCH  /ontology/entities/{entity}/{id}           部分更新
POST   /ontology/graph/traverse                   多跳查询
POST   /ontology/timeseries/{entity}/write        时序写入
POST   /ontology/timeseries/{entity}/query        时序聚合
POST   /ontology/actions/{action}                 执行 Action
GET    /ontology/meta                             返回已加载的本体元数据

OpenAPI 是泛化 schema(entity 走路径参数),不逐实体生成 typed 路由——这是刻意的取舍,见 AGENTS.md §4。

8. 集成到其他项目

交付形态是 pip 包,不是独立服务。 本适配器不启动 HTTP 服务、不占端口、没有自己的 procfile——它住在宿主进程里。

pip install git+ssh://git@gitlab.cnsjzn.com/donnyzhang/adapter_python_pg.git
pip install "adapter_python_pg[api] @ git+ssh://..."   # 需要 FastAPI 路由时

三种用法,同一个包:

  1. 纯库(最常见):宿主只调 Adapter 的语义接口,不挂 HTTP 路由,FastAPI 依赖都不用装。
  2. 挂路由:宿主已有 FastAPI 应用,app.include_router(build_router(adapter)) 把它当子路由,权限、中间件、鉴权沿用宿主那套。
  3. 嵌入式:本体 YAML 打进宿主仓库,adapter 作为宿主内部的数据层,外部完全不知道 PostgreSQL 存在。

核心依赖只有 psycopg / age / pydantic / pyyaml / cel-python;FastAPI 属于 [api] 可选依赖,装了才有路由能力。

9. 与存储的关系

底层依赖多模型 PostgreSQL:PostgreSQL 18 + TimescaleDB 2.31 + Apache AGE 1.7,文档/图/时序同库。

本机参考部署在 E:\code\ueos\devops\devos\server-resources\postgres-multimodel\(Dockerfile + compose + 初始化脚本)。

10. 非目标

  • 不做独立服务:不提供可独立部署的 HTTP 服务、不做多租户网关、不接管鉴权(那是宿主的事)
  • 不做通用 ORM(不复刻 SQLAlchemy 的表达式系统;复杂 SQL 走 raw() 逃生口)
  • 不做本体编辑器 / 可视化(那是前端的事)
  • 不做数据迁移与版本演进工具(本体版本变更先手工处理,后续再谈)
  • 不做多库联邦(一个 adapter 实例对一个 PostgreSQL 实例)
  • 不在 YAML 里执行任意 Python(expr 走受限表达式求值)

11. 路线

阶段 内容
v0.1 本体解析 + Schema Compiler + Repository(文档读写、约束校验)
v0.2 Links 落到 AGE、图遍历、路径查询
v0.3 时序实体与 hypertable、批量写入、连续聚合查询
v0.4 Rules 推理(派生关系/属性)、Functions 求值
v0.5 FastAPI 路由自动生成、Actions 运行时
v1.0 接口冻结、错误模型统一、文档与示例齐全