runic-ogm

name

runic-ogm

description

Expert guide for runic.ogm — a SQLModel-style, graph-native Python OGM for Cypher databases (FalkorDB, Neo4j, Memgraph, ArcadeDB, Apache AGE). Use whenever the user defines graph models (Node/Edge), maps fields, declares relationships, writes graph queries/traversals, or does session/repository CRUD with runic. Invoke for any code that imports from `runic.ogm`, uses `Node`, `Edge`, `Field`, `Relation`, `Session`, `Repository`, `select()`, or the query builder, and for any "how do I model/query this graph in runic" task. This is the OGM skill; for schema migrations use the `runic-migrate` skill instead.

runic.ogm — Graph OGM for Cypher Databases

runic.ogm is a lightweight, SQLModel-inspired OGM for property-graph databases. You declare nodes and edges as typed Python classes, then create, read, relate, and query them through a Session — without hand-writing Cypher. It targets FalkorDB first and also runs on Neo4j, Memgraph, ArcadeDB, and Apache AGE through a pluggable driver layer.

Three pillars, each with a runnable example and a reference:

Pillar	What it covers	Start here
Mapping	Node/Edge classes, Field(), defaults, PK, indexes, native types	examples/mapping.py
Relations	Relation(), lazy/eager loading, edge models, relate(), polymorphism	examples/relations.py
Query builder	select(), filters, traversal, aggregation, search	examples/query_builder.py

For the full API surface (every Field/Relation parameter, all Session, Repository, and QueryBuilder methods, drivers, exceptions) read references/api-reference.md. For task-oriented recipes and gotchas read references/cookbook.md.

The snippets below omit the # type: ignore / # noqa comments the repo's own examples carry. Those exist only because Field()/Relation() return Any and the descriptor comparison operators (User.age > 18) confuse some type checkers. The code is correct as written; add the ignores only if your checker complains.

---

Quick start

All runic.ogm imports come from runic.ogm directly. Session, Repository, select, and their methods (add, commit, flush, scalars, relate, unrelate, query, count, all_rows, etc.) are runic.ogm-native API — not from SQLAlchemy or any other ORM. Never import from runic.ogm.orm.* — the package root re-exports everything.

from runic.ogm import (
    Field, Node, Edge, Relation,
    Session, AsyncSession,
    Repository,
    select,
    count, avg, sum_,
)
from runic.ogm.driver.factory import create_driver

class User(Node, labels=["User"]):
    id: str = Field(primary_key=True)
    name: str
    email: str = Field(unique=True)
    active: bool = True

driver = create_driver("falkordb", host="localhost", port=6379, graph="app")

with Session(driver) as session:
    session.add(User(id="u1", name="Alice", email="[e-mail]"))
    session.commit()

    alice = session.get(User, "u1")     # read by primary key
    alice.name = "Alice B."             # mutation marks the entity dirty
    session.commit()                    # flushes the SET automatically

driver.close()

Embedded FalkorDB (no server, great for tests/examples):

from redislite import FalkorDB
from runic.ogm.driver.falkordb import FalkorDBDriver

db = FalkorDB(protocol=2)               # protocol=2 avoids a redis-py 8 issue
driver = FalkorDBDriver(db.select_graph("app"))

---

Mapping: defining models

A Node is a vertex; declare it with a labels list. An Edge is a relationship property model; declare it with a type string.

from datetime import datetime
from enum import StrEnum
from runic.ogm import Edge, Field, Node, Vector, GeoLocation

class Status(StrEnum):
    DRAFT = "draft"
    PUBLISHED = "published"

class Article(Node, labels=["Article"]):
    id: str = Field(primary_key=True)        # explicit PK
    title: str                               # bare annotation → required field
    summary: str | None = None               # optional → defaults to None
    status: Status                            # Enum → EnumConverter auto-applied
    published_at: datetime | None = None      # datetime → DatetimeConverter auto
    country: str = Field(interned=True)        # intern() dedup (FalkorDB)
    embedding: Vector | None = None           # vecf32() vector (FalkorDB)
    origin: GeoLocation | None = None         # point() geo (FalkorDB)
    views: int = Field(default=0)

class Authored(Edge, type="AUTHORED"):       # edge property model
    at: datetime
    primary: bool = Field(default=False)

Field declaration — two equivalent styles. A bare annotation (title: str) is auto-promoted to a Field. Use the explicit Field(...) form only when you need options: primary_key, unique, index, index_type, default, default_factory, converter, interned, generated. See the full parameter table in references/api-reference.md.

Best practices for mapping:

• Pick the primary key deliberately. Field(primary_key=True) is explicit and clearest. A field named id is treated as the PK by convention if none is marked. For DB-generated identifiers use Field(generated=True).
• Constructors are keyword-only. Always Article(id="a1", title="…"), never positional. Unknown kwargs and missing required fields raise TypeError.
• Let auto-converters work. datetime, Enum, Vector, and GeoLocation fields get the right TypeConverter automatically — don't pass converter= for these. Explicit converter= always wins if you need custom behavior.
• Native types are FalkorDB-specific. Vector (vecf32), GeoLocation (point), and interned=True (intern) are wrapped only on FalkorDB; other backends store the raw value. See examples/native_types.py.
• Indexes: Field(index=True) for a range index, Field(unique=True) for a uniqueness constraint, Field(index_type="FULLTEXT") / ="VECTOR" to back search. The declarations live on the model; actual index creation is done by runic.migrate's IndexManager (see the runic-migrate skill).

Polymorphism uses multi-label nodes with a shared primary_label. A query on the base class returns the correct concrete subtypes:

class Location(Node, labels=["Location"], primary_label="Location"):
    id: str
    title: str

class City(Location, labels=["Location", "City"], primary_label="Location"):
    population: int | None = None

Repository(session, Location).find_all() returns a mix of City, etc., each as its concrete type. Full example: examples/polymorphism.py.

---

Relations: connecting nodes

Declare a relationship with Relation(). relationship, direction, and target are required; edge_model attaches a property model.

from runic.ogm import Node, Relation

class User(Node, labels=["User"]):
    id: str = Field(primary_key=True)
    name: str
    # single related node
    manager: "User | None" = Relation(
        relationship="REPORTS_TO", direction="OUTGOING", target="User"
    )
    # collection, with edge properties
    articles: list["Article"] = Relation(
        relationship="AUTHORED", direction="OUTGOING",
        target="Article", edge_model="Authored",
    )
    # symmetric / undirected
    contacts: list["User"] = Relation(
        relationship="KNOWS", direction="BOTH", target="User"
    )

Reading relationships — lazy vs eager:

# Lazy (default): first attribute access runs a query. Needs a live session.
user = session.get(User, "u1")
for a in user.articles:        # triggers the load here
    ...

# Eager: fetch alongside the parent in one query — no later round-trips.
user = session.get(User, "u1", fetch=["articles"])

Mutating relationships — use relate() / unrelate() instead of raw Cypher. relate() uses MERGE semantics: it is idempotent and re-calling relate() with edge=EdgeModel(...) UPDATES the edge properties in place — no need for unrelate() + relate() just to update edge properties. Pass the class-level descriptor for type safety:

session.relate(user, User.articles, article,
               edge=Authored(at=datetime.now(UTC), primary=True))
# re-calling updates the edge in place — idempotent:
session.relate(user, User.articles, article,
               edge=Authored(at=datetime.now(UTC), primary=False))
session.unrelate(user, User.articles, article)   # remove the relationship

Best practices for relations:

• Default to lazy; reach for fetch= when you know you need the data, especially in loops (avoids N+1) or before detaching an entity.
• Async has no lazy loading. In AsyncSession, accessing an unloaded relation raises LazyLoadError — you must use await session.get(Cls, pk, fetch=[...]). relate() / unrelate() in async are coroutines (await session.relate(...)). The session-bound query builder is also async: await session.query(Cls).where(...).all(). See examples/async_session.py.
• Detached entities raise DetachedEntityError on lazy access. After the session closes or you expunge(), the data is gone — fetch eagerly first.
• Mirror two-sided relationships by declaring the same relationship type with OUTGOING on one class and INCOMING on the other; both views read the same edges. Use direction="BOTH" for genuinely undirected links.
• Edge properties belong on an Edge model named to match the type; read them with all_with_edges() (below). Full example: examples/relations.py.

---

Query builder: reading the graph

Build statements with select(Cls) (session-independent) and run them through the session. Filters come from comparing class-level field descriptors.

from runic.ogm import select

# list of entities
users = session.scalars(
    select(User).where(User.active == True).order_by(User.name).limit(20)
)

# one-or-none
alice = session.scalar(select(User).where(User.email == "[e-mail]"))

# count
n = session.count(select(User).where(User.active == True))

Execution methods (session takes a select(...) statement):

Method	Returns
session.scalars(stmt)	list[Entity]
session.scalar(stmt)	Entity | None
session.count(stmt)	int
session.all_rows(stmt)	list[dict] — for project() / aggregate()
session.all_with_edges(stmt)	list[tuple] — node/edge/node rows

session.query(Cls) returns a session-bound builder with terminal methods on it instead: .all(), .one(), .count(), .all_rows(), .all_with_edges(). Both styles are equivalent; select() is preferred for composable statements.

Filtering — descriptor operators produce filter expressions:

select(Product).where(Product.price > 100)
select(Product).where(Product.name.contains("Graph"))
select(Product).where(Product.sku.is_null())
select(Product).where(Product.id.in_(["p1", "p2"]))
# Boolean composition — parenthesize each operand:
select(Product).where((Product.category == "books") & (Product.active == True))
select(Product).where((Product.a == 1) | (Product.b == 2))
select(Product).where(~(Product.active == True))

Operators: ==, !=, <, <=, >, >=, .contains(), .startswith(), .endswith(), .matches() (regex), .is_null(), .is_not_null(), .in_(), .not_in_(). Multiple .where() calls are AND-combined. Refinements: .order_by(field, desc=True), .limit(n), .skip(n), .distinct().

Projection & aggregation return rows (dicts), read via all_rows():

from runic.ogm import avg, count, sum_

rows = session.all_rows(select(Product).project(Product.name, Product.price))
# keys are "n.name", "n.price"

summary = session.all_rows(
    select(Order).where(Order.status == "done").aggregate(
        count("*").as_("total"),
        sum_(Order.amount).as_("revenue"),
        avg(Order.amount).as_("avg"),
    )
)
# group by a FIELD — pass "alias.property" (default node alias is "n"):
per_city = session.all_rows(
    select(User).aggregate(count("*").as_("n_users"), group_by="n.city")
)
# → [{"n.city": "Berlin", "n_users": 3}, {"n.city": "Paris", "n_users": 2}, ...]

# group by the whole node — pass the bare alias (handy after a traversal):
per_user = session.all_rows(
    select(User).alias("u").traverse(User.orders).aggregate(
        sum_(Order.amount).as_("revenue"), group_by="u"
    )
)

group_by keeps its argument verbatim in the RETURN, and Cypher groups on any non-aggregated term. Pass "alias.property" (e.g. "n.city") to group by a field — the result key is then that same string ("n.city"); pass a bare alias to group by the whole node. Helpers: count, avg, sum_, min_, max_, collect (each with .as_(); count/collect accept distinct=True).

Traversal — alias nodes, hop across relationships, filter by alias:

# Alice's active friends-of-friends who authored a post
posts = session.scalars(
    select(User).alias("u").where(User.id == "alice")
    .traverse(User.friends).alias("f")
    .where(User.active == True, on="f")
    .traverse(User.authored).alias("post")
    .return_target("post")
)

• .traverse(Cls.rel, edge_alias="e", optional=True) — one hop. optional=True (default) is a left join; optional=False is a required (inner) join.
• When filtering on an edge property, use optional=False — an OPTIONAL MATCH + WHERE nullifies non-matching rows instead of dropping them.
• .repeat(Cls.rel, min_hops=1, max_hops=3) — variable-length path.
• .with_("alias") — pipeline stages. .return_target("alias"), .return_nodes(...), .return_edge("e") — choose result columns.
• Filter on an aliased node or edge with .where(expr, on="alias").

Edge properties come back via all_with_edges():

rows = session.all_with_edges(
    select(User).alias("u").where(User.id == "alice")
    .traverse(User.rated_movies, edge_alias="r").alias("m")
    .where(Rated.score >= 9.0, on="r")
    .return_nodes("u", "m").return_edge("r")
)
for user, edge, movie in rows:    # (User, Rated, Movie) tuples
    ...

Search (FalkorDB) — needs the relevant index created first:

session.fulltext_search(Article, query="graph databases").where(
    Article.published == True
).limit(10).all()

session.vector_search(
    Article, field=Article.embedding, vector=[0.1, 0.2, 0.3, 0.4], k=3
).all()

Inspect without executing: cypher, params = select(...).build(). Reach for this when debugging a query or explaining what runic generates.

---

Sessions & repositories

• Lifecycle: use with Session(driver) as session:. add() / add_all() stage inserts; delete() stages removals. flush() writes pending changes without ending the logical transaction; commit() flushes then clears tracking; rollback() discards pending state. Setting any attribute on a loaded entity marks it dirty for the next commit.
• Identity map: within one session, get() returns the same instance for a given (type, pk). rollback() discards pending state; expire()/refresh() re-read from the graph; expunge() detaches.
• Repository wraps a session + model for collection reads: Repository(session, User).find_all(skip=0, limit=50), .find_all_by_ids(ids), .count(), .exists(pk). Pagination is offset-based via skip/limit (there is no Page object).
• Custom repositories subclass Repository[T] and add methods using self.query() (the bound builder) or the raw-Cypher helpers:
  • self.cypher(q, params, returns=Cls) — maps rows to entities
  • self.cypher(q, params, returns=int) — maps to a scalar; write=True for mutations
  • self.cypher_one(q, params, returns=int) — first mapped row; use for single scalars
  • self.cypher_raw(q, params) — returns GraphResult with .columns (list of names) and .rows (list of value lists)

class ArticleRepository(Repository[Article]):
    def by_author(self, author: str) -> list[Article]:
        return self.cypher(
            "MATCH (a:Article {author: $a}) RETURN a", {"a": author}, returns=Article
        )

    def count_by_status(self, status: str) -> int:
        return self.cypher_one(
            "MATCH (a:Article {status: $s}) RETURN count(a)", {"s": status}, returns=int
        )

    def stats(self):
        raw = self.cypher_raw(
            "MATCH (a:Article) RETURN a.author AS author, count(a) AS n", {}
        )
        return [dict(zip(raw.columns, row)) for row in raw.rows]

See references/api-reference.md for every method signature and references/cookbook.md for async, multi-backend, performance, and error-handling recipes.

---

Key rules

• Prefer relate()/unrelate() and the query builder over raw Cypher; drop to session.execute() / repo.cypher() only for what the builder can't express.
• Use fetch=[...] to avoid N+1 in loops and always in async code.
• optional=False whenever you filter on a traversed edge's properties.
• Vector, GeoLocation, and interned are FalkorDB-only native types; on other backends the raw value is stored. Fulltext and vector search also work on Neo4j and Memgraph but require pre-created named indexes (see cookbook multi-backend table); they are absent on ArcadeDB and Apache AGE.
• Multi-backend driver params differ. FalkorDB uses graph= (the graph name); Neo4j/Memgraph/ArcadeDB use database=, username=, password=. Neo4j defaults to encrypted=True. Example:

  create_driver("neo4j", host="localhost", port=7687,
                database="neo4j", username="neo4j", password="…")  # encrypted=True
  

  Pass the driver to Session(driver) — models and queries are unchanged.
• Indexes are declared on models but created by runic.migrate — keep schema changes in migrations, not application code.