Relationships

runic.orm models relationships as first-class graph edges. This page covers lazy loading, eager loading, polymorphic hierarchies, and edge properties.

Declaring a relationship

Use Relation() with relationship, direction, and target. Property fields use Field() — the two are intentionally separate:

from runic.orm import Field, Node, Relation

class Company(Node, labels=["Company"]):
    id: str = Field(primary_key=True, generated=True)
    name: str = Field(index=True)

class Person(Node, labels=["Person"]):
    id: str = Field(primary_key=True, generated=True)
    name: str = Field(index=True)
    # single outgoing relationship
    company: Company | None = Relation(
        relationship="WORKS_FOR",
        direction="OUTGOING",
        target="Company",
    )
    # collection
    reports: list["Person"] = Relation(
        relationship="MANAGES",
        direction="OUTGOING",
        target="Person",
    )

See also

examples/orm/03_relationships_and_edges.py

Full runnable example: declaring relationships, lazy vs eager loading, relate() / unrelate(), and edge-property queries.

Lazy loading (default)

Relationship fields are not loaded when the entity is fetched. Accessing the attribute triggers a graph query on first read:

with Session(graph) as session:
    person = session.get(Person, "alice")
    company = person.company     # ← one Cypher query here

Note

In an AsyncSession, lazy loading raises LazyLoadError because __get__ cannot await. Use fetch=[...] instead.

Eager loading

Pass fetch=["field_name", ...] to session.get() or any Repository read:

with Session(graph) as session:
    # Single entity
    person = session.get(Person, "alice", fetch=["company"])
    company = person.company    # ← no extra query

with Session(graph) as session:
    repo = Repository(session, Person)
    # Entire collection
    people = repo.find_all(fetch=["company"])

The Mapper builds a single Cypher query with one OPTIONAL MATCH per entry in fetch. Related entities are also registered in the session’s identity map.

See also

examples/orm/02_polymorphic_locations.py

Multi-label hierarchy (Location Country, City, Restaurant) with subtype resolution and repository queries.

Polymorphic hierarchies

Nodes can carry multiple labels and form inheritance chains. Declare a primary_label to control which label is used in MATCH statements:

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

class Country(Location, labels=["Location", "Country"], primary_label="Location"):
    iso_code: str = Field(unique=True)

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

Querying via the parent class returns all subtypes; each node is decoded to its most specific registered class:

with Session(graph) as session:
    repo = Repository(session, Location)
    all_locs = repo.find_all()
    # returns a mix of Country, City, etc. — type-resolved per node
    for loc in all_locs:
        print(type(loc).__name__, loc.title)

Mutating relationships

Use relate() and unrelate() to create, update, or remove relationships without writing Cypher:

with Session(graph) as session:
    alice = session.get(User, "alice")
    company = session.get(Company, "acme")

    # Create (or update) the relationship — MERGE semantics
    session.relate(alice, "company", company)

    # Remove the relationship
    session.unrelate(alice, "company", company)

relate() is idempotent: calling it a second time does not duplicate the edge. The cached field value on the source entity is invalidated after each mutation so the next access re-fetches from the graph.

For async sessions the same methods are available as coroutines:

async with AsyncSession(graph) as session:
    alice = await session.get(User, "alice")
    company = await session.get(Company, "acme")
    await session.relate(alice, "company", company)

Edge properties

When a relationship carries its own properties, declare an Edge subclass and pass it via edge_model:

from runic.orm import Edge, Field, Node, Relation

class InvitationEdge(Edge, type="INVITED_TO"):
    role: str = Field()
    status: str = Field()
    invited_at: str = Field()          # ISO-8601
    accepted_at: str | None = Field(default=None)

class User(Node, labels=["User"]):
    id: str = Field()
    invited_trips: list["Trip"] = Relation(
        relationship="INVITED_TO",
        direction="OUTGOING",
        target="Trip",
        edge_model=InvitationEdge,
    )

Pass an Edge instance to relate() to write properties onto the relationship. Because relate() uses MERGE, calling it again with updated values will overwrite the existing properties:

with Session(graph) as session:
    user = session.get(User, "alice")
    trip = session.get(Trip, "paris-2026")

    # Create — or update if the edge already exists
    session.relate(
        user,
        "invited_trips",
        trip,
        edge=InvitationEdge(
            role="owner",
            status="accepted",
            invited_at="2026-01-01T00:00:00",
        ),
    )

Read edge properties back with a custom Cypher query in your Repository:

from runic.orm import Repository

class UserRepository(Repository[User]):
    def get_invitation(self, user_id: str, trip_id: str) -> dict | None:
        return self.cypher_one(
            """
            MATCH (u:User {id: $uid})-[e:INVITED_TO]->(t:Trip {id: $tid})
            RETURN e.role AS role, e.status AS status,
                   e.invited_at AS invited_at, e.accepted_at AS accepted_at
            """,
            {"uid": user_id, "tid": trip_id},
            returns=dict,
        )

Cascade saves

Set cascade=True on a Relation to automatically stage related entities when the owning entity is added:

class Person(Node, labels=["Person"]):
    id: str = Field()
    name: str = Field()
    company: Company | None = Relation(
        relationship="WORKS_FOR",
        direction="OUTGOING",
        target="Company",
        cascade=True,
    )

with Session(graph) as session:
    company = Company(id="acme", name="Acme")
    person = Person(id="alice", name="Alice", company=company)
    session.add(person)     # also stages company
    session.commit()
    assert company.id is not None