rust/lancedb
1
0
Fork 0
mirror of https://github.com/lancedb/lancedb.git synced 2026-11-13 14:16:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: manage property graphs defined over tables

Browse Source 
Replaces #4322, keeping only its property graph defined over tables.

A property graph maps node labels and relationship types to existing tables of its namespace, in the shape of SQL/PGQ's `CREATE PROPERTY GRAPH`. It stores only that definition: a query reads the tables as nodes and edges through SQL `GRAPH_TABLE`, as they are when it runs.

- **Rust:** `Connection::create_property_graph` takes a `PropertyGraphDefinition` of `NodeTable`s and `EdgeTable`s. `describe_property_graph`, `drop_property_graph` and `list_property_graphs` manage it.
- **Python:** `lancedb.graph` has `NodeTable`, `EdgeTable` and `Endpoint`. Sync and async connections gain `create_property_graph`, `open_property_graph`, `drop_property_graph` and `list_property_graphs`, and a `PropertyGraph` handle describes its graph.
- **Support:** remote connections only. Local connections raise `NotImplementedError` in Python and return `NotSupported` in Rust.

```python
import lancedb
from lancedb.graph import EdgeTable, Endpoint, NodeTable

db = lancedb.connect("db://demo", api_key="...")
db.create_property_graph(
    "social",
    nodes=[NodeTable("person", key="person_id", label="Person")],
    edges=[
        EdgeTable(
            "knows",
            label="KNOWS",
            source=Endpoint("src_id", references=("person", "person_id")),
            destination=Endpoint("dst_id", references=("person", "person_id")),
        )
    ],
)
db.execute_query("""
    SELECT * FROM GRAPH_TABLE (social
        MATCH (p:Person)-[:KNOWS]->(f:Person) WHERE p.age > 30
        COLUMNS (p.name AS person, f.name AS friend))
""").read_all()
```
This commit is contained in:
Jack Ye
2026-10-31 08:25:06 +08:00
committed by GitHub
parent d316cbbed4
commit 29691719fa
14 changed files with 1735 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/python/lancedb/__init__.py
+3
View File
@@ -48,6 +48,7 @@ from .materialized_view import (
MaterializedViewDefinition,
)
from .view import ViewDescription as ViewDescription
from .graph import PropertyGraph, PropertyGraphDescription
from .table import AsyncTable, CompactionOptions, Table
from .types import BaseTokenizerType
from ._lancedb import Session
@@ -603,6 +604,8 @@ __all__ = [
"MaterializedView",
"MaterializedViewDefinition",
"ViewDescription",
"PropertyGraph",
"PropertyGraphDescription",
"connect",
"connect_async",
"tokenize",
python/python/lancedb/_lancedb.pyi
+19
View File
@@ -225,6 +225,25 @@ class Connection(object):
page_token: Optional[str] = None,
page_limit: Optional[int] = None,
) -> NameListing: ...
async def create_property_graph(
self,
name: str,
definition_json: str,
namespace_path: Optional[List[str]] = None,
) -> str: ...
async def describe_property_graph(
self, name: str, namespace_path: Optional[List[str]] = None
) -> str: ...
async def drop_property_graph(
self, name: str, namespace_path: Optional[List[str]] = None
) -> None: ...
def list_property_graphs(
self,
namespace_path: Optional[List[str]] = None,
*,
page_token: Optional[str] = None,
page_limit: Optional[int] = None,
) -> NameListing: ...
def list_jobs(
self, *, page_token: Optional[str] = None, page_limit: Optional[int] = None
) -> JobListing: ...
python/python/lancedb/db.py
+216
View File
@@ -59,6 +59,13 @@ from lance_namespace.errors import NamespaceNotEmptyError, TableNotFoundError
from . import __version__
from ._lancedb import connect as lancedb_connect # type: ignore
from .functions import FunctionVersion, UdfDefinition
from .graph import (
AsyncPropertyGraph,
EdgeTable,
NodeTable,
PropertyGraph,
_definition_json,
)
from .job import AsyncJob, Job, _typed_job
from .storage_integration import SecretReference, StorageIntegrationInfo
from .sql import AsyncQuery as AsyncSqlQuery
@@ -1222,6 +1229,99 @@ class DBConnection(EnforceOverrides):
"View operations are not supported for this connection type"
)
def create_property_graph(
self,
name: str,
nodes: Sequence[NodeTable],
edges: Sequence[EdgeTable] = (),
*,
namespace_path: Optional[List[str]] = None,
) -> PropertyGraph:
"""Create a property graph over tables in its namespace.
Each [NodeTable][lancedb.graph.NodeTable] names the table holding one
label's nodes and its key column, and each
[EdgeTable][lancedb.graph.EdgeTable] the table holding one label's
edges and the node keys its source and destination columns reference
-- the shape of SQL/PGQ's ``CREATE PROPERTY GRAPH``. The graph stores
nothing else: a query reads its tables through SQL ``GRAPH_TABLE``, as
they are when it runs.
The database checks the definition against the tables now, so one they
cannot satisfy is refused here rather than at the first query. A name
another property graph holds is an error. Local connections raise
``NotImplementedError``.
>>> import lancedb
>>> from lancedb.graph import EdgeTable, Endpoint, NodeTable
>>> db = lancedb.connect("db://my_database") # doctest: +SKIP
>>> graph = db.create_property_graph(
... "social",
... nodes=[NodeTable("person", key="person_id", label="Person")],
... edges=[
... EdgeTable(
... "knows",
... label="KNOWS",
... source=Endpoint("src_id", references=("person", "person_id")),
... destination=Endpoint(
... "dst_id", references=("person", "person_id")
... ),
... )
... ],
... ) # doctest: +SKIP
>>> graph.describe().nodes[0].properties # doctest: +SKIP
['person_id', 'name']
"""
raise NotImplementedError(
"Property graph operations are not supported for this connection type"
)
def open_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> PropertyGraph:
"""Open the property graph named ``name``.
The graph is described to check that it exists, so a name no property
graph holds raises [HttpError][lancedb.remote.errors.HttpError] with
``status_code`` 404. Local connections raise ``NotImplementedError``.
"""
raise NotImplementedError(
"Property graph operations are not supported for this connection type"
)
def drop_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> None:
"""Drop a property graph and wait for its definition to be deleted.
The tables it reads are untouched: a property graph holds no rows of
its own. Local connections raise ``NotImplementedError``.
"""
raise NotImplementedError(
"Property graph operations are not supported for this connection type"
)
def list_property_graphs(
self,
*,
namespace_path: Optional[List[str]] = None,
page_token: Optional[str] = None,
page_limit: Optional[int] = None,
) -> Listing[str]:
"""The names of the property graphs in one namespace.
Names only; a definition comes from
[PropertyGraph.describe][lancedb.graph.PropertyGraph.describe]. Local
connections raise ``NotImplementedError``.
Returns a lazy iterator. ``page_limit`` limits each request, not the total;
``page_token`` resumes from a saved token. Requests and errors occur during
iteration. See [pagination state][lancedb.listing.AsyncListing].
"""
raise NotImplementedError(
"Property graph operations are not supported for this connection type"
)
def open_job(self, job_id: str) -> Job:
"""Open a server-side job by id, returning a handle with its record
already populated.
@@ -2285,6 +2385,61 @@ class LanceDBConnection(DBConnection):
)
)
@override
def create_property_graph(
self,
name: str,
nodes: Sequence[NodeTable],
edges: Sequence[EdgeTable] = (),
*,
namespace_path: Optional[List[str]] = None,
) -> PropertyGraph:
return PropertyGraph(
LOOP.run(
self._conn.create_property_graph(
name, nodes, edges, namespace_path=namespace_path
)
)
)
@override
def open_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> PropertyGraph:
return PropertyGraph(
LOOP.run(
self._conn.open_property_graph(name, namespace_path=namespace_path)
)
)
@override
def drop_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> None:
LOOP.run(self._conn.drop_property_graph(name, namespace_path=namespace_path))
@override
def list_property_graphs(
self,
*,
namespace_path: Optional[List[str]] = None,
page_token: Optional[str] = None,
page_limit: Optional[int] = None,
) -> Listing[str]:
"""List property graphs.
Returns a lazy iterator. ``page_limit`` limits each request, not the total;
``page_token`` resumes from a saved token. Requests and errors occur during
iteration. See [pagination state][lancedb.listing.AsyncListing].
"""
return Listing(
self._conn.list_property_graphs(
namespace_path=namespace_path,
page_token=page_token,
page_limit=page_limit,
)
)
@override
def list_jobs(
self, *, page_token: Optional[str] = None, page_limit: Optional[int] = None
@@ -3597,6 +3752,67 @@ class AsyncConnection(object):
)
)
async def create_property_graph(
self,
name: str,
nodes: Sequence[NodeTable],
edges: Sequence[EdgeTable] = (),
*,
namespace_path: Optional[List[str]] = None,
) -> AsyncPropertyGraph:
"""Create a property graph over tables in its namespace.
See
[DBConnection.create_property_graph][lancedb.DBConnection.create_property_graph].
"""
definition = _definition_json(nodes, edges)
graph = AsyncPropertyGraph(self._inner, name, namespace_path)
await self._inner.create_property_graph(name, definition, graph.namespace_path)
return graph
async def open_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> AsyncPropertyGraph:
"""Open the property graph named ``name``, describing it to check that
it exists.
See
[DBConnection.open_property_graph][lancedb.DBConnection.open_property_graph].
"""
graph = AsyncPropertyGraph(self._inner, name, namespace_path)
await graph.describe()
return graph
async def drop_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> None:
"""Drop a property graph and wait for its definition to be deleted.
The tables it reads are untouched.
"""
await self._inner.drop_property_graph(name, list(namespace_path or []))
def list_property_graphs(
self,
*,
namespace_path: Optional[List[str]] = None,
page_token: Optional[str] = None,
page_limit: Optional[int] = None,
) -> AsyncListing[str]:
"""The names of the property graphs in one namespace.
Returns a lazy iterator. ``page_limit`` limits each request, not the total;
``page_token`` resumes from a saved token. Requests and errors occur during
iteration. See [pagination state][lancedb.listing.AsyncListing].
"""
return AsyncListing(
self._inner.list_property_graphs(
namespace_path=namespace_path,
page_token=page_token,
page_limit=page_limit,
)
)
def list_jobs(
self, *, page_token: Optional[str] = None, page_limit: Optional[int] = None
) -> AsyncListing[JobInfo]:
python/python/lancedb/graph.py
+256
View File
@@ -0,0 +1,256 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright The LanceDB Authors
"""Property graphs: graphs defined over tables a database already holds.
A property graph stores no rows. Its definition says which tables of its
namespace hold nodes and which hold edges: each node table's key and label
(:class:`NodeTable`), and each edge table's label and the node keys its source
and destination columns reference (:class:`EdgeTable`) -- the shape of
SQL/PGQ's ``CREATE PROPERTY GRAPH``. A query reads the tables through SQL
``GRAPH_TABLE``, as they are when it runs.
See ``DBConnection.create_property_graph``.
"""
from __future__ import annotations
import json
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Sequence, Tuple
from .background_loop import LOOP
if TYPE_CHECKING:
from ._lancedb import Connection as LanceDbConnection
@dataclass(frozen=True)
class NodeTable:
"""A table whose rows are the nodes of one label."""
table: str
"""The table, in the graph's namespace."""
key: str
"""The column that identifies a node."""
label: str
"""The label its nodes carry."""
properties: Optional[List[str]] = None
"""The columns exposed as node properties; every column when ``None``. A
created graph reports the list it resolved to, the key among them."""
@dataclass(frozen=True)
class Endpoint:
"""One end of an edge: a column of the edge table and the node key it holds."""
column: str
"""The edge table's column."""
references: Tuple[str, str]
"""The node table and its key column, as ``(table, column)``."""
@dataclass(frozen=True)
class EdgeTable:
"""A table whose rows are the edges of one label."""
table: str
"""The table, in the graph's namespace."""
label: str
"""The label its edges carry."""
source: Endpoint
"""The column naming each edge's source node."""
destination: Endpoint
"""The column naming each edge's destination node."""
properties: Optional[List[str]] = None
"""The columns exposed as edge properties; every column when ``None``. A
created graph reports the list it resolved to, both endpoint columns among
them."""
@dataclass(frozen=True)
class PropertyGraphDescription:
"""What a database records about one property graph."""
name: str
"""The graph's name within its namespace."""
namespace_path: List[str]
"""The namespace holding the graph; empty is the root namespace."""
nodes: List[NodeTable]
"""The node tables, each with the properties create resolved."""
edges: List[EdgeTable]
"""The edge tables, each with the properties create resolved."""
def _definition_json(nodes: Sequence[NodeTable], edges: Sequence[EdgeTable]) -> str:
"""The definition a create request carries, refusing elements of the wrong
type before any request is made."""
nodes, edges = list(nodes), list(edges)
for element, expected in [
*((node, NodeTable) for node in nodes),
*((edge, EdgeTable) for edge in edges),
]:
if not isinstance(element, expected):
raise TypeError(
"A property graph's nodes are NodeTable and its edges EdgeTable, "
f"not {type(element).__name__}"
)
return json.dumps(
{
"nodes": [_node_table_to_json(node) for node in nodes],
"edges": [_edge_table_to_json(edge) for edge in edges],
}
)
def _description_from_json(text: str) -> PropertyGraphDescription:
described = json.loads(text)
return PropertyGraphDescription(
name=described["name"],
namespace_path=list(described.get("namespace", [])),
nodes=[_node_table_from_json(node) for node in described["nodes"]],
edges=[_edge_table_from_json(edge) for edge in described.get("edges", [])],
)
def _properties_to_json(
encoded: Dict[str, Any], properties: Optional[List[str]]
) -> None:
# Absent means every column, which an empty list would not.
if properties is None:
return
if isinstance(properties, str):
raise TypeError("properties is a list of column names, not a string")
encoded["properties"] = list(properties)
def _node_table_to_json(node: NodeTable) -> Dict[str, Any]:
encoded: Dict[str, Any] = {
"table": node.table,
"key": node.key,
"label": node.label,
}
_properties_to_json(encoded, node.properties)
return encoded
def _edge_table_to_json(edge: EdgeTable) -> Dict[str, Any]:
encoded: Dict[str, Any] = {
"table": edge.table,
"label": edge.label,
"source": _endpoint_to_json(edge.source),
"destination": _endpoint_to_json(edge.destination),
}
_properties_to_json(encoded, edge.properties)
return encoded
def _endpoint_to_json(endpoint: Endpoint) -> Dict[str, Any]:
if not isinstance(endpoint, Endpoint):
raise TypeError(
"An edge's source and destination are Endpoint, "
f"not {type(endpoint).__name__}"
)
table, column = endpoint.references
return {
"column": endpoint.column,
"references": {"table": table, "column": column},
}
def _node_table_from_json(node: Dict[str, Any]) -> NodeTable:
return NodeTable(
table=node["table"],
key=node["key"],
label=node["label"],
properties=node.get("properties"),
)
def _edge_table_from_json(edge: Dict[str, Any]) -> EdgeTable:
return EdgeTable(
table=edge["table"],
label=edge["label"],
source=_endpoint_from_json(edge["source"]),
destination=_endpoint_from_json(edge["destination"]),
properties=edge.get("properties"),
)
def _endpoint_from_json(endpoint: Dict[str, Any]) -> Endpoint:
references = endpoint["references"]
return Endpoint(
column=endpoint["column"],
references=(references["table"], references["column"]),
)
class AsyncPropertyGraph:
"""A handle on one property graph: its name and namespace, and the
connection it is described through.
Obtained from ``AsyncConnection.create_property_graph`` or
``AsyncConnection.open_property_graph``. The graph is queried through SQL
``GRAPH_TABLE``, not through this handle.
"""
def __init__(
self,
inner: "LanceDbConnection",
name: str,
namespace_path: Optional[List[str]] = None,
):
self._inner = inner
self._name = name
self._namespace_path = list(namespace_path or [])
def __repr__(self) -> str:
return (
f"AsyncPropertyGraph(name={self._name!r}, "
f"namespace_path={self._namespace_path!r})"
)
@property
def name(self) -> str:
"""The graph's name within its namespace."""
return self._name
@property
def namespace_path(self) -> List[str]:
"""The namespace holding the graph; empty is the root namespace."""
return list(self._namespace_path)
async def describe(self) -> PropertyGraphDescription:
"""The graph's node and edge tables, each with the properties create
resolved."""
return _description_from_json(
await self._inner.describe_property_graph(self._name, self._namespace_path)
)
class PropertyGraph:
"""Synchronous variant of
[AsyncPropertyGraph][lancedb.graph.AsyncPropertyGraph]."""
def __init__(self, graph: AsyncPropertyGraph):
self._async = graph
def __repr__(self) -> str:
return (
f"PropertyGraph(name={self.name!r}, namespace_path={self.namespace_path!r})"
)
@property
def name(self) -> str:
"""The graph's name within its namespace."""
return self._async.name
@property
def namespace_path(self) -> List[str]:
"""The namespace holding the graph; empty is the root namespace."""
return self._async.namespace_path
def describe(self) -> PropertyGraphDescription:
"""The graph's node and edge tables, each with the properties create
resolved."""
return LOOP.run(self._async.describe())
python/python/lancedb/remote/db.py
+56
View File
@@ -38,6 +38,7 @@ from ..common import DATA
from lancedb.listing import Listing, NamespaceListing, TableListing
from ..db import DBConnection, LOOP
from ..functions import FunctionVersion, UdfDefinition
from ..graph import EdgeTable, NodeTable, PropertyGraph
from ..job import AsyncJob, Job
from ..sql import Query as SqlQuery
from ..sql import QueryDescription, QueryParameters
@@ -1187,6 +1188,61 @@ class RemoteDBConnection(DBConnection):
)
)
@override
def create_property_graph(
self,
name: str,
nodes: Sequence[NodeTable],
edges: Sequence[EdgeTable] = (),
*,
namespace_path: Optional[List[str]] = None,
) -> PropertyGraph:
return PropertyGraph(
LOOP.run(
self._conn.create_property_graph(
name, nodes, edges, namespace_path=namespace_path
)
)
)
@override
def open_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> PropertyGraph:
return PropertyGraph(
LOOP.run(
self._conn.open_property_graph(name, namespace_path=namespace_path)
)
)
@override
def drop_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> None:
LOOP.run(self._conn.drop_property_graph(name, namespace_path=namespace_path))
@override
def list_property_graphs(
self,
*,
namespace_path: Optional[List[str]] = None,
page_token: Optional[str] = None,
page_limit: Optional[int] = None,
) -> Listing[str]:
"""List property graphs.
Returns a lazy iterator. ``page_limit`` limits each request, not the total;
``page_token`` resumes from a saved token. Requests and errors occur during
iteration. See [pagination state][lancedb.listing.AsyncListing].
"""
return Listing(
self._conn.list_property_graphs(
namespace_path=namespace_path,
page_token=page_token,
page_limit=page_limit,
)
)
@override
def list_jobs(
self, *, page_token: Optional[str] = None, page_limit: Optional[int] = None
python/python/tests/test_remote_db.py
+224
View File
@@ -2,6 +2,8 @@
# SPDX-FileCopyrightText: Copyright The LanceDB Authors
from concurrent.futures import ThreadPoolExecutor
import contextlib
import copy
from dataclasses import replace
from datetime import timedelta
import http.server
import io
@@ -24,6 +26,14 @@ import lancedb
import numpy as np
from lancedb.conftest import MockNonNormTextEmbeddingFunction, MockTextEmbeddingFunction
from lancedb.embeddings import EmbeddingFunctionConfig, EmbeddingFunctionRegistry
from lancedb.graph import (
AsyncPropertyGraph,
EdgeTable,
Endpoint,
NodeTable,
PropertyGraph,
PropertyGraphDescription,
)
from lancedb.query import (
AsyncQuery,
ColumnOrdering,
@@ -4583,3 +4593,217 @@ def test_remote_job_handle_reports_its_own_detail():
"limit": 500,
"filter": "state = 'claim_complete'",
}
def _property_graph_handler(calls, respond):
"""A mock route handler recording ``(method, path, body)`` and answering
with the ``(status, body)`` that ``respond(path, body)`` returns, as JSON."""
def handler(request):
length = int(request.headers.get("Content-Length") or 0)
raw = request.rfile.read(length) if length else b""
body = json.loads(raw) if raw else None
calls.append((request.command, request.path, body))
status, payload = respond(request.path, body)
request.send_response(status)
request.send_header("Content-Type", "application/json")
request.end_headers()
request.wfile.write(json.dumps(payload).encode())
return handler
SOCIAL_NODES = [
NodeTable("person", key="person_id", label="Person", properties=["name"])
]
SOCIAL_EDGES = [
EdgeTable(
"knows",
label="KNOWS",
source=Endpoint("src_id", references=("person", "person_id")),
destination=Endpoint("dst_id", references=("person", "person_id")),
)
]
# The create body for the tables above. The edge table leaves its properties
# out, which means every column.
SOCIAL_DEFINITION = {
"nodes": [
{
"table": "person",
"key": "person_id",
"label": "Person",
"properties": ["name"],
}
],
"edges": [
{
"table": "knows",
"label": "KNOWS",
"source": {
"column": "src_id",
"references": {"table": "person", "column": "person_id"},
},
"destination": {
"column": "dst_id",
"references": {"table": "person", "column": "person_id"},
},
}
],
}
# Create resolves each table's properties, the key and endpoint columns always
# among them.
RESOLVED_NODES = [replace(SOCIAL_NODES[0], properties=["person_id", "name"])]
RESOLVED_EDGES = [replace(SOCIAL_EDGES[0], properties=["src_id", "dst_id", "since"])]
def _social_description(namespace):
described = copy.deepcopy(SOCIAL_DEFINITION)
described["nodes"][0]["properties"] = RESOLVED_NODES[0].properties
described["edges"][0]["properties"] = RESOLVED_EDGES[0].properties
return {"name": "social", "namespace": namespace, **described}
def test_property_graph_crud_addresses_its_own_routes():
# The definition is the create body and describe posts an empty object.
# An accepted drop hands back its cleanup job, which drop waits on.
calls = []
def respond(path, body):
if path.startswith("/v1/namespace/analytics/property_graph/list"):
if "page_token=p2" in path:
return 200, {"property_graphs": ["payments"]}
return 200, {"property_graphs": ["social"], "page_token": "p2"}
if path.endswith("/drop"):
return 202, {"job_id": "j1-pg-drop"}
if path == "/v1/jobs/describe":
return 200, {"job_id": body["job_id"], "job_state": "DONE"}
return 200, _social_description(["analytics"])
with mock_lancedb_connection(_property_graph_handler(calls, respond)) as db:
graph = db.create_property_graph(
"social", SOCIAL_NODES, SOCIAL_EDGES, namespace_path=["analytics"]
)
assert isinstance(graph, PropertyGraph)
assert (graph.name, graph.namespace_path) == ("social", ["analytics"])
assert graph.describe() == PropertyGraphDescription(
name="social",
namespace_path=["analytics"],
nodes=RESOLVED_NODES,
edges=RESOLVED_EDGES,
)
opened = db.open_property_graph("social", namespace_path=["analytics"])
assert (opened.name, opened.namespace_path) == ("social", ["analytics"])
graphs = db.list_property_graphs(namespace_path=["analytics"], page_limit=1)
assert list(graphs) == ["social", "payments"]
db.drop_property_graph("social", namespace_path=["analytics"])
assert calls == [
("POST", "/v1/property_graph/analytics$social/create", SOCIAL_DEFINITION),
("POST", "/v1/property_graph/analytics$social/describe", {}),
("POST", "/v1/property_graph/analytics$social/describe", {}),
("GET", "/v1/namespace/analytics/property_graph/list?limit=1", None),
(
"GET",
"/v1/namespace/analytics/property_graph/list?page_token=p2&limit=1",
None,
),
("POST", "/v1/property_graph/analytics$social/drop", None),
("POST", "/v1/jobs/describe", {"job_id": "j1-pg-drop"}),
]
@pytest.mark.asyncio
async def test_async_property_graph_handles():
calls = []
def respond(path, body):
if path.endswith("/property_graph/list"):
return 200, {"property_graphs": ["social"]}
if path.endswith("/drop"):
return 200, {}
return 200, _social_description([])
async with mock_lancedb_connection_async(
_property_graph_handler(calls, respond)
) as db:
graph = await db.create_property_graph("social", SOCIAL_NODES)
assert isinstance(graph, AsyncPropertyGraph)
described = await graph.describe()
assert (described.namespace_path, described.nodes) == ([], RESOLVED_NODES)
opened = await db.open_property_graph("social")
assert isinstance(opened, AsyncPropertyGraph)
assert [name async for name in db.list_property_graphs()] == ["social"]
await db.drop_property_graph("social")
assert calls == [
(
"POST",
"/v1/property_graph/social/create",
{"nodes": SOCIAL_DEFINITION["nodes"], "edges": []},
),
("POST", "/v1/property_graph/social/describe", {}),
("POST", "/v1/property_graph/social/describe", {}),
("GET", "/v1/namespace/$/property_graph/list", None),
("POST", "/v1/property_graph/social/drop", None),
]
def test_a_missing_property_graph_is_not_found():
# Opening describes the graph, so a name no graph holds fails there.
calls = []
def respond(path, body):
return 404, {"error": "property graph 'social' not found"}
with mock_lancedb_connection(_property_graph_handler(calls, respond)) as db:
with pytest.raises(HttpError) as missing:
db.open_property_graph("social")
assert missing.value.status_code == 404
with pytest.raises(HttpError) as missing:
db.drop_property_graph("social")
assert missing.value.status_code == 404
assert calls == [
("POST", "/v1/property_graph/social/describe", {}),
("POST", "/v1/property_graph/social/drop", None),
]
def test_a_property_graph_takes_node_and_edge_tables():
# A definition of the wrong shape is refused before any request is made.
calls = []
with mock_lancedb_connection(
_property_graph_handler(calls, lambda path, body: (200, {}))
) as db:
with pytest.raises(TypeError, match="nodes are NodeTable"):
db.create_property_graph("social", SOCIAL_EDGES)
with pytest.raises(TypeError, match="not NodeTable"):
db.create_property_graph("social", SOCIAL_NODES, SOCIAL_NODES)
with pytest.raises(TypeError, match="are Endpoint, not tuple"):
db.create_property_graph(
"social",
SOCIAL_NODES,
[replace(SOCIAL_EDGES[0], source=("person", "person_id"))],
)
with pytest.raises(TypeError, match="not a string"):
db.create_property_graph(
"social", [replace(SOCIAL_NODES[0], properties="name")]
)
assert calls == []
def test_local_connections_refuse_property_graphs(tmp_path):
db = lancedb.connect(tmp_path)
refused = "Property graph operations are not supported"
with pytest.raises(NotImplementedError, match=refused):
db.create_property_graph("social", SOCIAL_NODES, SOCIAL_EDGES)
with pytest.raises(NotImplementedError, match=refused):
db.open_property_graph("social")
with pytest.raises(NotImplementedError, match=refused):
list(db.list_property_graphs())
with pytest.raises(NotImplementedError, match=refused):
db.drop_property_graph("social")
python/src/connection.rs
+74
View File
@@ -24,6 +24,7 @@ use lancedb::{
connection::NamespaceClientPushdownOperation,
database::namespace::LanceNamespaceDatabase,
database::{CreateTableMode, Database, ReadConsistency},
graph::PropertyGraphDefinition,
};
use pyo3::{
Bound, FromPyObject, Py, PyAny, PyRef, PyResult, Python,
@@ -1116,6 +1117,79 @@ impl Connection {
))
}
/// Create a property graph from its definition's JSON, and return its
/// description's JSON. The Python layer names the fields; the JSON keeps
/// the binding free of classes kept in step with the Rust structs.
#[pyo3(signature = (name, definition_json, namespace_path=None))]
pub fn create_property_graph(
self_: PyRef<'_, Self>,
name: String,
definition_json: String,
namespace_path: Option<Vec<String>>,
) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.get_inner()?.clone();
let namespace_path = namespace_path.unwrap_or_default();
future_into_py(self_.py(), async move {
let definition = PropertyGraphDefinition::from_json(&definition_json).infer_error()?;
inner
.create_property_graph(name, &definition, &namespace_path)
.await
.infer_error()?
.to_json()
.infer_error()
})
}
/// The JSON description of the property graph named `name`.
#[pyo3(signature = (name, namespace_path=None))]
pub fn describe_property_graph(
self_: PyRef<'_, Self>,
name: String,
namespace_path: Option<Vec<String>>,
) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.get_inner()?.clone();
let namespace_path = namespace_path.unwrap_or_default();
future_into_py(self_.py(), async move {
inner
.describe_property_graph(name, &namespace_path)
.await
.infer_error()?
.to_json()
.infer_error()
})
}
#[pyo3(signature = (name, namespace_path=None))]
pub fn drop_property_graph(
self_: PyRef<'_, Self>,
name: String,
namespace_path: Option<Vec<String>>,
) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.get_inner()?.clone();
let namespace_path = namespace_path.unwrap_or_default();
future_into_py(self_.py(), async move {
inner
.drop_property_graph(name, &namespace_path)
.await
.infer_error()
})
}
#[pyo3(signature = (namespace_path=None, *, page_token=None, page_limit=None))]
pub fn list_property_graphs(
&self,
namespace_path: Option<Vec<String>>,
page_token: Option<String>,
page_limit: Option<u32>,
) -> PyResult<crate::listing::NameListing> {
Ok(crate::listing::NameListing::new(
self.get_inner()?.list_property_graphs(
&namespace_path.unwrap_or_default(),
crate::listing::options(page_token, page_limit),
),
))
}
#[pyo3(signature = (*, page_token=None, page_limit=None))]
pub fn list_jobs(
&self,