from __future__ import annotations
from typing import Any, Protocol, runtime_checkable
from runic.migrate.introspect import LiveSchema, SchemaSnapshot
@runtime_checkable
class GraphAdapter(Protocol):
"""Protocol all graph-database adapters must satisfy.
The runic core depends only on this interface — no FalkorDB or any other
concrete database client leaks into shared code.
Note: ``LiveSchema`` (returned by ``read_live_schema``) is currently parsed
from FalkorDB's ``CALL db.indexes()`` / ``CALL db.constraints()`` output in
``runic.migrate.introspect``. A future adapter must override ``read_live_schema``
and may supply its own introspection logic.
"""
@property
def name(self) -> str: ...
# Normalised query execution (satisfies runic.orm.operations._Executor)
def execute(self, cypher: str, params: dict[str, Any]) -> Any: ...
# Low-level query execution
def run_query(self, query: str, params: dict | None = None) -> Any: ...
def run_ro_query(self, query: str) -> Any: ...
# Sibling adapter for a different graph/database on the same connection
def fork(self, graph_name: str) -> GraphAdapter: ...
# Version tracking
def get_version(self) -> list[str]: ...
def set_version(self, revisions: list[str]) -> None: ...
# Schema introspection
def read_live_schema(self) -> LiveSchema: ...
def introspect_schema(self) -> SchemaSnapshot: ...
# Schema DDL
def create_range_index(
self, label: str, prop: str, *, rel: bool = False
) -> None: ...
def drop_range_index(self, label: str, prop: str, *, rel: bool = False) -> None: ...
def create_fulltext_index(
self,
label: str,
*props: str,
language: str | None = None,
stopwords: list[str] | None = None,
) -> None: ...
def drop_fulltext_index(self, label: str, *props: str) -> None: ...
def create_vector_index(
self,
label: str,
prop: str,
dimension: int,
similarity: str,
*,
m: int = 16,
ef_construction: int = 200,
ef_runtime: int = 10,
) -> None: ...
def drop_vector_index(self, label: str, prop: str) -> None: ...
def create_constraint(
self, kind: str, entity: str, label: str, props: list[str]
) -> None: ...
def drop_constraint(
self, kind: str, entity: str, label: str, props: list[str]
) -> None: ...
# Lifecycle
def delete_graph(self) -> None: ...
# Snapshots
def supports_snapshots(self) -> bool: ...
def snapshot(self, snap_name: str) -> None: ...
def restore_snapshot(self, snap_name: str) -> None: ...
def snapshot_exists(self, snap_name: str) -> bool: ...
# Checksum & attribution tracking
def get_checksums(self) -> dict[str, str]: ...
def set_checksum(
self, rev_id: str, checksum: str, installed_by: str | None = None
) -> None: ...
def get_installed_by(self) -> dict[str, str]: ...
[docs]
def create_adapter(backend: str, **kwargs: Any) -> GraphAdapter:
"""Instantiate a named adapter from keyword arguments.
Two connection variants are supported for ``"falkordb"``:
**URL variant** — credentials embedded in the connection string::
create_adapter(
"falkordb", url="falkor://:mypassword@localhost:6379", graph_name="my_graph"
)
**Params variant** — explicit host/port/auth kwargs::
create_adapter(
"falkordb",
host="localhost",
port=6379,
username="myuser",
password="mypassword",
graph_name="my_graph",
)
"""
if backend == "falkordb":
from runic.migrate.adapters.falkordb import FalkorDBAdapter
graph_name = kwargs["graph_name"]
if "url" in kwargs:
return FalkorDBAdapter.from_url(
kwargs["url"],
graph_name,
username=kwargs.get("username"),
password=kwargs.get("password"),
)
return FalkorDBAdapter.from_params(
graph_name,
host=kwargs.get("host", "localhost"),
port=int(kwargs.get("port", 6379)),
username=kwargs.get("username"),
password=kwargs.get("password"),
)
if backend == "arcadedb":
from runic.migrate.adapters.arcadedb import ArcadeDBAdapter
return ArcadeDBAdapter.from_params(
database=kwargs["database"],
host=kwargs.get("host", "localhost"),
port=int(kwargs.get("port", 7687)),
username=kwargs.get("username", "root"),
password=kwargs.get("password", "playwithdata"),
)
if backend == "age":
from runic.migrate.adapters.age import AGEAdapter
return AGEAdapter.from_params(
graph=kwargs["graph"],
host=kwargs.get("host", "localhost"),
port=int(kwargs.get("port", 5432)),
database=kwargs.get("database", "postgres"),
username=kwargs.get("username", "postgres"),
password=kwargs.get("password", ""),
)
if backend == "neo4j":
from runic.migrate.adapters.neo4j import Neo4jAdapter
return Neo4jAdapter.from_params(
database=kwargs["database"],
host=kwargs.get("host", "localhost"),
port=int(kwargs.get("port", 7687)),
username=kwargs.get("username", "neo4j"),
password=kwargs.get("password", ""),
encrypted=bool(kwargs.get("encrypted", True)),
)
if backend == "memgraph":
from runic.migrate.adapters.memgraph import MemgraphAdapter
return MemgraphAdapter.from_params(
database=kwargs.get("database", "memgraph"),
host=kwargs.get("host", "localhost"),
port=int(kwargs.get("port", 7687)),
username=kwargs.get("username", ""),
password=kwargs.get("password", ""),
encrypted=bool(kwargs.get("encrypted", False)),
)
raise KeyError(
f"Unknown adapter backend {backend!r}. "
"Supported: 'falkordb', 'arcadedb', 'age', 'neo4j', 'memgraph'"
)
__all__ = ["GraphAdapter", "create_adapter"]