feat: 自研 agtype 解析与 psycopg3 loader

- graph/agtype.py:解析 ::vertex / ::edge / ::path 后缀,返回 Vertex/Edge/Path
- register_agtype() 直接查 ag_catalog.agtype 的 OID 注册 loader(TypeInfo.fetch 取不到)
- 18 条单元测试,样本全部从真实 AGE 库抓取;7 条畸形输入用例
- tests/test_agtype_live.py:连真库的集成测试,默认跳过(ADAPTER_TEST_DSN)
- AGENTS.md 技术栈表:图驱动改为自研解析,注明不用 apache-age-python 的原因
This commit is contained in:
Donny committed 2026-11-05 02:30:11 +08:00
1 parent ace505cc6b
commit f042452fe6
5 files changed
+443 -14

No files matched your search

+104
View File
@@ -0,0 +1,104 @@
"""agtype 解析测试。
下面的原始文本全部是从真实 AGE 库(pg-multimodel / demo_graph)里 `select v` 抓下来的,
不是手写的近似值——格式一旦对不上,AGE 升级就会在这里先报错。
"""
from __future__ import annotations
import pytest
from adapter_python_pg.graph.agtype import (
AGTypeError,
Edge,
Path,
Vertex,
parse_agtype,
)
VERTEX_RAW = '{"id": 844424930131971, "label": "Person", "properties": {"name": "Alice"}}::vertex'
EDGE_RAW = (
'{"id": 1125899906842626, "label": "KNOWS", "end_id": 844424930131972, '
'"start_id": 844424930131971, "properties": {"since": 2021}}::edge'
)
PATH_RAW = (
'[{"id": 844424930131971, "label": "Person", "properties": {"name": "Alice"}}::vertex, '
'{"id": 1125899906842626, "label": "KNOWS", "end_id": 844424930131972, '
'"start_id": 844424930131971, "properties": {"since": 2021}}::edge, '
'{"id": 844424930131972, "label": "Person", "properties": {"name": "Bob"}}::vertex]::path'
)
SCALARS_RAW = '["Alice", 42, 3.5, true, null, {"k": "v", "n": 1}]'
def test_parse_vertex() -> None:
value = parse_agtype(VERTEX_RAW)
assert isinstance(value, Vertex)
assert value.id == 844424930131971
assert value.label == "Person"
assert value.properties == {"name": "Alice"}
assert value.get("name") == "Alice"
assert value.get("missing", "fallback") == "fallback"
def test_parse_edge() -> None:
value = parse_agtype(EDGE_RAW)
assert isinstance(value, Edge)
assert value.id == 1125899906842626
assert value.label == "KNOWS"
assert value.start_id == 844424930131971
assert value.end_id == 844424930131972
assert value.properties == {"since": 2021}
assert value.get("since") == 2021
def test_parse_path() -> None:
value = parse_agtype(PATH_RAW)
assert isinstance(value, Path)
assert len(value.elements) == 3
assert len(value.vertices) == 2
assert len(value.edges) == 1
assert value.start.properties["name"] == "Alice"
assert value.end.properties["name"] == "Bob"
assert value.edges[0].label == "KNOWS"
def test_parse_scalars_and_nested() -> None:
value = parse_agtype(SCALARS_RAW)
assert value == ["Alice", 42, 3.5, True, None, {"k": "v", "n": 1}]
def test_parse_accepts_bytes_and_memoryview() -> None:
assert isinstance(parse_agtype(VERTEX_RAW.encode()), Vertex)
assert isinstance(parse_agtype(memoryview(VERTEX_RAW.encode())), Vertex)
def test_empty_list() -> None:
assert parse_agtype("[]") == []
def test_empty_properties_defaults_to_dict() -> None:
value = parse_agtype('{"id": 1, "label": "X", "properties": {}}::vertex')
assert isinstance(value, Vertex)
assert value.properties == {}
def test_unknown_annotation_is_stripped() -> None:
# 形如 ::numeric 的注解对调用方没有意义,直接去掉
assert parse_agtype("42::numeric") == 42
@pytest.mark.parametrize(
("raw", "reason"),
[
('{"id": 1, "label": "X"}::vertex', "缺少 properties"),
('{"label": "X", "properties": {}}::vertex', "缺少 id"),
('{"id": 1, "label": "X", "properties": {}}::edge', "边缺少 start_id"),
('["a", "b"::vertex', "数组未闭合"),
('["a" "b"]', "数组分隔符非法"),
("[] trailing", "尾部有多余内容"),
('{"id": 1, "label": "X", "properties": {}}::path', "path 应为数组"),
],
)
def test_malformed_input_raises(raw: str, reason: str) -> None:
with pytest.raises(AGTypeError):
parse_agtype(raw)
+73
View File
@@ -0,0 +1,73 @@
"""agtype loader 的集成测试:连真实 AGE 库,验证 psycopg3 侧的类型注册。
默认跳过(需要环境变量 `ADAPTER_TEST_DSN`):
ADAPTER_TEST_DSN="host=127.0.0.1 port=5432 dbname=multimodel user=postgres password=..." \
uv run pytest -m integration
"""
from __future__ import annotations
import os
import pytest
from adapter_python_pg.graph.agtype import Edge, Path, Vertex, register_agtype
DSN = os.environ.get("ADAPTER_TEST_DSN")
pytestmark = [
pytest.mark.integration,
pytest.mark.skipif(not DSN, reason="未设置 ADAPTER_TEST_DSN,跳过集成测试"),
]
CYPHER_VERTEX = """
SELECT v FROM cypher('demo_graph', $$ MATCH (p:Person {name:'Alice'}) RETURN p $$) AS (v agtype)
"""
CYPHER_EDGE = """
SELECT e FROM cypher('demo_graph', $$
MATCH (:Person {name:'Alice'})-[r:KNOWS]->(:Person) RETURN r
$$) AS (e agtype)
"""
CYPHER_PATH = """
SELECT p FROM cypher('demo_graph', $$
MATCH p=(:Person {name:'Alice'})-[:KNOWS]->(:Person) RETURN p LIMIT 1
$$) AS (p agtype)
"""
@pytest.fixture()
def conn() -> object:
psycopg = pytest.importorskip("psycopg")
with psycopg.connect(DSN) as connection:
register_agtype(connection)
with connection.cursor() as cur:
cur.execute('SET search_path = ag_catalog, "$user", public')
yield connection
def _one(conn: object, sql: str) -> object:
with conn.cursor() as cur: # type: ignore[attr-defined]
cur.execute(sql)
return cur.fetchone()[0]
def test_vertex_roundtrip(conn: object) -> None:
value = _one(conn, CYPHER_VERTEX)
assert isinstance(value, Vertex)
assert value.label == "Person"
assert value.properties["name"] == "Alice"
def test_edge_roundtrip(conn: object) -> None:
value = _one(conn, CYPHER_EDGE)
assert isinstance(value, Edge)
assert value.label == "KNOWS"
assert value.start_id != value.end_id
def test_path_roundtrip(conn: object) -> None:
value = _one(conn, CYPHER_PATH)
assert isinstance(value, Path)
assert len(value.elements) == 3
assert value.start.properties["name"] == "Alice"