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:

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="alice@example.com"))
    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 == "alice@example.com"))

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

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

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.