ORM API Reference¶

runic.orm is a lightweight graph ORM that maps Python classes to FalkorDB nodes and edges. It follows a SQLAlchemy-style architecture: models → metadata → session → repository.

runic.orm.core — Models & Fields¶

class runic.orm.core.models.Node[source]¶

Bases: object

Base class for graph nodes.

Subclass with labels and optional primary_label class keywords:

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

__init__ is generated automatically from declared Field descriptors. Setting any field on an existing instance marks it dirty (_dirty = True), which the Mapper interprets as a MERGE/SET on the next flush.

class runic.orm.core.models.Edge[source]¶

Bases: object

Base class for graph edge property models.

Subclass with a type class keyword:

class InvitationEdge(Edge, type="INVITED_TO"):
    role: str = Field(required=True)

Edge instances carry the same _new/_dirty lifecycle flags as Node.

runic.orm.core.descriptors.Field(*, default=MISSING, default_factory=None, init=True, kw_only=True, index=False, index_type=None, unique=False, required=False, primary_key=False, converter=None, generated=False, interned=False)[source]¶

Declare a property field on a Node or Edge.

Returns a FieldDescriptor typed as Any so that name: str = Field() is accepted by type checkers without error.

Set interned=True to store the value via FalkorDB’s intern() function, which deduplicates repeated strings (e.g. country names, status codes, tags) by keeping a single shared copy in the database.

Example:

class Person(Node, labels=["Person"]):
    id: str = Field(primary_key=True)
    name: str = Field()
    age: int | None = Field(default=None)
    country: str = Field(interned=True)
    email: str = Field(index=True, unique=True)

Parameters:

default (Any)
default_factory (Callable[[], Any] | None)
init (bool)
kw_only (bool)
index (bool)
index_type (Literal['FULLTEXT', 'VECTOR'] | None)
unique (bool)
required (bool)
primary_key (bool)
converter (TypeConverter | None)
generated (bool)
interned (bool)

Return type:

Any

runic.orm.core.descriptors.Relation(*, relationship, direction, target, edge_model=None, cascade=False, lazy=True, default=None, default_factory=None, init=True)[source]¶

Declare a relationship field on a Node.

Returns a FieldDescriptor typed as Any so that company: Company | None = Relation(...) is accepted by type checkers.

Example:

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

Parameters:

relationship (str)
direction (Literal['OUTGOING', 'INCOMING', 'BOTH'])
target (str | type)
edge_model (str | type | None)
cascade (bool)
lazy (bool)
default (Any)
default_factory (Callable[[], Any] | None)
init (bool)

Return type:

Any

class runic.orm.core.descriptors.FieldDescriptor(*, default=MISSING, default_factory=None, init=True, kw_only=True, index=False, index_type=None, unique=False, required=False, primary_key=False, relationship=None, direction=None, target=None, edge_model=None, cascade=False, lazy=True, converter=None, generated=False, interned=False)[source]¶

Bases: object

Descriptor backing both Field() and Relation() declarations.

Behaves as a data descriptor: values are stored per-instance in __dict__. Writing to an instance attribute via this descriptor sets instance._dirty = True, which the Mapper uses to detect changes on persistent entities.

Query expression operators¶

When accessed at the class level (User.name, Rated.score), the descriptor returns itself. The comparison operators defined below produce FilterExpr objects for use with where():

session.query(User).where(User.name == "Alice")
session.query(User).where(User.age > 18)
session.query(User).where(User.bio.contains("graph"))
session.query(User).where(User.deleted_at.is_null())

__hash__ is kept as object.__hash__ so descriptors remain hashable and can appear in sets/dicts used internally.

property name: str¶: Public read-only accessor for the attribute name set by __set_name__.

property has_default: bool¶: True if a default value or factory is configured.

get_default()[source]¶

Return the default value, calling default_factory if set.

Return type:: Any

property field_name: str¶: Public accessor for the descriptor’s attribute name.

property owner: type | None¶: Public accessor for the class that owns this descriptor.

contains(value)[source]¶

Return a CONTAINS filter: prop CONTAINS $value.

Parameters:: value (str)
Return type:: Any

startswith(value)[source]¶

Return a STARTS WITH filter: prop STARTS WITH $value.

Parameters:: value (str)
Return type:: Any

endswith(value)[source]¶

Return an ENDS WITH filter: prop ENDS WITH $value.

Parameters:: value (str)
Return type:: Any

matches(pattern)[source]¶

Return a regex filter: prop =~ $pattern.

Parameters:: pattern (str)
Return type:: Any

is_null()[source]¶

Return an IS NULL filter.

Return type:: Any

is_not_null()[source]¶

Return an IS NOT NULL filter.

Return type:: Any

in_(values)[source]¶

Return an IN filter: prop IN $values.

Parameters:: values (list[Any])
Return type:: Any

not_in_(values)[source]¶

Return a NOT IN filter: NOT prop IN $values.

Parameters:: values (list[Any])
Return type:: Any

Parameters:

default (Any)
default_factory (Callable[[], Any] | None)
init (bool)
kw_only (bool)
index (bool)
index_type (Literal['FULLTEXT', 'VECTOR'] | None)
unique (bool)
required (bool)
primary_key (bool)
relationship (str | None)
direction (Literal['OUTGOING', 'INCOMING', 'BOTH'] | None)
target (str | type | None)
edge_model (str | type | None)
cascade (bool)
lazy (bool)
converter (TypeConverter | None)
generated (bool)
interned (bool)

class runic.orm.core.descriptors.FieldInfo(name, field, is_collection=False)[source]¶

Bases: object

Pairs a field name with its FieldDescriptor; used during __init__ generation.

Parameters:

name (str)
field (FieldDescriptor)
is_collection (bool)

runic.orm.core — MetaData¶

class runic.orm.core.metadata.MetaData[source]¶

Bases: object

Registry for all Node and Edge subclasses.

Populated automatically when Node/Edge subclasses are defined via __init_subclass__. Also provides forward-reference resolution (string targets on relationship Fields) after all models are imported.

register_node(cls)[source]¶

Parameters:: cls (type)
Return type:: None

register_edge(cls)[source]¶

Parameters:: cls (type)
Return type:: None

get_node_meta(cls)[source]¶

Return metadata for a Node class, or None if not registered.

Parameters:: cls (type)
Return type:: NodeMeta | None

get_edge_meta(cls)[source]¶

Return metadata for an Edge class, or None if not registered.

Parameters:: cls (type)
Return type:: EdgeMeta | None

resolve_node_by_label(label)[source]¶

Look up a Node by its primary label string.

Parameters:: label (str)
Return type:: NodeMeta | None

resolve_edge_by_type(edge_type)[source]¶

Look up an Edge by its type string.

Parameters:: edge_type (str)
Return type:: EdgeMeta | None

resolve_target(target)[source]¶

Resolve a string forward reference to its registered Node/Edge class.

Parameters:: target (str | type | None)
Return type:: type | None

finalize()[source]¶

Resolve all string forward references in relationship Fields.

Call once after all model modules have been imported. String targets on relationship Fields are replaced with the actual class objects.

Return type:: None

all_nodes()[source]¶

Return metadata for all registered Node subclasses.

Return type:: list[NodeMeta]

all_edges()[source]¶

Return metadata for all registered Edge subclasses.

Return type:: list[EdgeMeta]

snapshot()[source]¶

Capture the current registry state for later restore.

Return type:: _MetaSnapshot

restore(snap)[source]¶

Restore the registry to a prior snapshot (used in test fixtures).

Parameters:: snap (_MetaSnapshot)
Return type:: None

clear()[source]¶

Remove all registrations (use in tests only).

Return type:: None

class runic.orm.core.metadata.NodeMeta(cls, labels, primary_label, fields, pk_field_name=None)[source]¶

Bases: object

Metadata snapshot for a registered Node subclass.

Parameters:

cls (type)
labels (list[str])
primary_label (str)
fields (list[FieldInfo])
pk_field_name (str | None)

class runic.orm.core.metadata.EdgeMeta(cls, edge_type, fields)[source]¶

Bases: object

Metadata snapshot for a registered Edge subclass.

Parameters:

cls (type)
edge_type (str)
fields (list[FieldInfo])

runic.orm.core.metadata.get_metadata()[source]¶

Return the global MetaData singleton.

Return type:: MetaData

runic.orm.core — Type Converters¶

class runic.orm.core.types.TypeConverter[source]¶

Bases: ABC

Interface for encoding/decoding custom Python types to/from graph values.

Optionally, set cypher_fn to name a FalkorDB Cypher function that wraps the parameter reference when writing to the graph (e.g. "vecf32" → vecf32($field)).

abstractmethod to_graph(value)[source]¶

Convert a Python value to a graph-compatible representation.

Parameters:: value (Any)
Return type:: Any

abstractmethod from_graph(value)[source]¶

Convert a graph value back to the Python type.

Parameters:: value (Any)
Return type:: Any

class runic.orm.core.types.DatetimeConverter[source]¶

Bases: TypeConverter

Converts between Python datetime objects and ISO-8601 strings.

FalkorDB stores datetimes as strings; this converter handles the round-trip.

to_graph(value)[source]¶

Convert a Python value to a graph-compatible representation.

Parameters:: value (Any)
Return type:: Any

from_graph(value)[source]¶

Convert a graph value back to the Python type.

Parameters:: value (Any)
Return type:: Any

class runic.orm.core.types.EnumConverter(enum_class)[source]¶

Bases: TypeConverter

Converts between Python Enum members and their string values.

Stores the enum’s .value in the graph and reconstructs on load.

Parameters:: enum_class (type[Enum])

to_graph(value)[source]¶

Convert a Python value to a graph-compatible representation.

Parameters:: value (Any)
Return type:: Any

from_graph(value)[source]¶

Convert a graph value back to the Python type.

Parameters:: value (Any)
Return type:: Any

class runic.orm.core.types.Vector(iterable=(), /)[source]¶

Bases: list

A typed list of floats representing a graph embedding vector.

Use as an annotation on a Node field to store and query embeddings:

class Article(Node, labels=["Article"]):
    id: str = Field(primary_key=True)
    embedding: Vector = Field(index=True, index_type="VECTOR")

FalkorDB stores vectors via vecf32(), preserving 32-bit float precision.

class runic.orm.core.types.VectorConverter[source]¶

Bases: TypeConverter

Converts between Python Vector (list of floats) and FalkorDB’s vecf32 format.

Emits vecf32($field) in Cypher via cypher_fn = "vecf32".

to_graph(value)[source]¶

Convert a Python value to a graph-compatible representation.

Parameters:: value (Any)
Return type:: Any

from_graph(value)[source]¶

Convert a graph value back to the Python type.

Parameters:: value (Any)
Return type:: Any

class runic.orm.core.types.GeoLocation(latitude, longitude)[source]¶

Bases: object

A geographic point with latitude and longitude.

Maps to FalkorDB’s native point() type:

class Store(Node, labels=["Store"]):
    id: str = Field(primary_key=True)
    location: GeoLocation = Field()

FalkorDB round-trips this as point({latitude: ..., longitude: ...}).

Parameters:

latitude (float)
longitude (float)

class runic.orm.core.types.GeoLocationConverter[source]¶

Bases: TypeConverter

Converts between GeoLocation and FalkorDB’s point() dict format.

Emits point($field) in Cypher via cypher_fn = "point". FalkorDB returns points as {"latitude": ..., "longitude": ...} dicts.

to_graph(value)[source]¶

Convert a Python value to a graph-compatible representation.

Parameters:: value (Any)
Return type:: Any

from_graph(value)[source]¶

Convert a graph value back to the Python type.

Parameters:: value (Any)
Return type:: Any

runic.orm.session — Session¶

class runic.orm.session.session.Session(graph, mapper=None, *, log_cypher=False)[source]¶

Bases: object

Sync unit-of-work manager.

Owns all mutations (add, delete), single-entity lookup (get), identity map, and flush/commit lifecycle. Repositories hold a session reference and delegate writes and PK lookups to it.

FalkorDB transaction model: single GRAPH.QUERY is fully atomic. Multi-query uses sequential individual queries (no native pipeline in the Python client). rollback() discards un-flushed pending/deleted sets only; cannot undo writes already sent to the graph.

Parameters:

graph (Any)
mapper (Mapper | None)
log_cypher (bool)

add(entity)[source]¶

Parameters:: entity (Any)
Return type:: None

add_all(entities)[source]¶

Batch add.

Parameters:: entities (list[Any])
Return type:: None

delete(entity)[source]¶

Mark a persistent entity for DETACH DELETE on next flush.

Raises DetachedEntityError if the entity is not known to this session.

Parameters:: entity (Any)
Return type:: None

property mapper: Mapper¶: Return the Mapper used by this session.

property rel_loader: RelationshipLoader¶: Return the RelationshipLoader used by this session.

register_or_get(entity)[source]¶

Used by Repository reads to deduplicate against entities already loaded in this session (fulfilling the identity-map guarantee).

Parameters:: entity (Any)
Return type:: Any

get(cls, pk, fetch=None)[source]¶

Return entity from identity map or query graph; None if not found.

Pass fetch=["rel_name", ...] to eager-load relationship fields in the same Cypher query using OPTIONAL MATCH.

Parameters:

cls (type)
pk (Any)
fetch (list[str] | None)

Return type:

Any | None

load_relationship(entity, field_name)[source]¶

Load a lazy relationship field and cache the result on the entity.

Called by Field._trigger_lazy_load when a _NOT_LOADED sentinel is accessed on an entity that is attached to this session. Writes directly to entity.__dict__ to bypass the dirty-tracking descriptor.

Parameters:

entity (Any)
field_name (str)

Return type:

Any

flush()[source]¶

Execute all pending/dirty/deleted entities against the graph.

Does not clear the identity map. Each entity write is a separate graph.query() call. Entities with generated=True IDs are handled individually so the returned ID can be assigned before continuing.

Return type:: None

commit()[source]¶

flush() then clear the pending/deleted tracking sets.

Return type:: None

rollback()[source]¶

Discard un-flushed pending/deleted sets; expire all persistent entities.

Cannot undo writes already sent to the graph.

Return type:: None

expire(entity)[source]¶

Invalidate cached attributes; they will be reloaded on next refresh.

Parameters:: entity (Any)
Return type:: None

refresh(entity)[source]¶

Immediately re-query the entity from the graph and update in-place.

Parameters:: entity (Any)
Return type:: None

expunge(entity)[source]¶

Remove entity from session (→ detached); no graph action.

Parameters:: entity (Any)
Return type:: None

expunge_all()[source]¶

Expunge all tracked entities.

Return type:: None

relate(source, field_name, target, edge=None)[source]¶

Create or update a relationship between source and target.

Uses MERGE semantics: if the relationship already exists its edge properties are updated; if not, it is created. Pass an Edge model instance as edge to write properties on the relationship itself.

field_name may be a plain string or the class-level descriptor attribute (e.g. User.invited_trips) for type-safe call sites.

The cached value of the relation field on source is invalidated after the write so the next access re-fetches fresh data from the graph.

Parameters:

source (Any)
field_name (str | FieldDescriptor)
target (Any)
edge (Any | None)

Return type:

None

unrelate(source, field_name, target)[source]¶

Delete the relationship between source and target.

field_name may be a plain string or the class-level descriptor attribute (e.g. User.invited_trips) for type-safe call sites.

The cached value of the relation field on source is invalidated after the write so the next access re-fetches fresh data from the graph.

Parameters:

source (Any)
field_name (str | FieldDescriptor)
target (Any)

Return type:

None

execute(cypher, params=None, write=False)[source]¶

Execute raw Cypher; returns QueryResult; no entity mapping.

Parameters:

cypher (str)
params (dict[str, Any] | None)
write (bool)

Return type:

Any

query(cls)[source]¶

Return a QueryBuilder for cls.

This is the primary entry point for the fluent query builder API:

users = (
    session.query(User)
    .where(User.active == True)
    .order_by(User.name)
    .limit(20)
    .all()
)

Parameters:: cls (type[Any]) – A registered Node subclass.
Return type:: QueryBuilder[cls]

fulltext_search(cls, *, query, fields=None)[source]¶

Return a FulltextQueryBuilder for cls.

Uses FalkorDB’s CALL db.idx.fulltext.queryNodes() procedure. The node label must have a fulltext index created.

Parameters:

cls (type[Any]) – A registered Node subclass with at least one field with index_type="FULLTEXT".
query (str) – The fulltext search string.
fields (list[str] | None) – Optional list of field names to search (informational; the procedure uses the index it finds for the label).

Return type:

Any

Example

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

vector_search(cls, *, field, vector, k=10)[source]¶

Return a VectorQueryBuilder for cls.

Performs a K-Nearest-Neighbour search using FalkorDB’s HNSW index.

Parameters:

cls (type[Any]) – A registered Node subclass.
field (Any) – The FieldDescriptor of the Vector field to search (e.g. Document.embedding).
vector (list[float]) – The query embedding as a list of floats.
k (int) – Number of nearest neighbours to return (default 10).

Return type:

Any

Example

similar = (
    session.vector_search(
        Document, field=Document.embedding, vector=my_vec, k=5
    )
    .where(Document.active == True)
    .all()
)

close()[source]¶

expunge_all() and release the graph connection.

Return type:: None

class runic.orm.session.async_session.AsyncSession(graph, mapper=None, *, log_cypher=False)[source]¶

Bases: object

Async unit-of-work manager; mirrors Session with async methods.

Use as an async context manager:

async with AsyncSession(graph) as session:
    alice = await session.get(Person, "alice-id")
    alice.email = "new@example.com"
    await session.commit()

Lazy relationship loading is not supported in async context because Field.__get__ cannot await. Use fetch=[...] on get() instead.

Parameters:

graph (Any)
mapper (Mapper | None)
log_cypher (bool)

add(entity)[source]¶

Parameters:: entity (Any)
Return type:: None

add_all(entities)[source]¶

Batch add.

Parameters:: entities (list[Any])
Return type:: None

delete(entity)[source]¶

Mark a persistent entity for DETACH DELETE on next flush.

Parameters:: entity (Any)
Return type:: None

property mapper: Mapper¶: Return the Mapper used by this session.

property rel_loader: RelationshipLoader¶: Return the RelationshipLoader used by this session.

register_or_get(entity)[source]¶

Parameters:: entity (Any)
Return type:: Any

async get(cls, pk, fetch=None)[source]¶

Return entity from identity map or query graph asynchronously.

Pass fetch=["rel_name", ...] to eager-load relationships in the same Cypher query using OPTIONAL MATCH.

Parameters:

cls (type)
pk (Any)
fetch (list[str] | None)

Return type:

Any | None

load_relationship(entity, field_name)[source]¶

Raise LazyLoadError; lazy loading is not supported in async sessions.

Access entity.rel_field from within an async context manager triggers this via Field._trigger_lazy_load. Use fetch=[field_name] on get() for eager loading instead.

Parameters:

entity (Any)
field_name (str)

Return type:

Any

async flush()[source]¶

Execute all pending/dirty/deleted entities against the graph.

Return type:: None

async commit()[source]¶

flush() then clear the pending/deleted tracking sets.

Return type:: None

async rollback()[source]¶

Discard un-flushed pending/deleted sets; expire all persistent entities.

Return type:: None

expire(entity)[source]¶

Invalidate cached attributes.

Parameters:: entity (Any)
Return type:: None

async refresh(entity)[source]¶

Immediately re-query the entity from the graph.

Parameters:: entity (Any)
Return type:: None

expunge(entity)[source]¶

Remove entity from session; no graph action.

Parameters:: entity (Any)
Return type:: None

expunge_all()[source]¶

Expunge all tracked entities.

Return type:: None

async relate(source, field_name, target, edge=None)[source]¶

Create or update a relationship between source and target.

field_name may be a plain string or the class-level descriptor attribute (e.g. User.invited_trips) for type-safe call sites.

The cached value of the relation field on source is invalidated after the write so the next access re-fetches fresh data from the graph.

Parameters:

source (Any)
field_name (str | FieldDescriptor)
target (Any)
edge (Any | None)

Return type:

None

async unrelate(source, field_name, target)[source]¶

Delete the relationship between source and target.

field_name may be a plain string or the class-level descriptor attribute (e.g. User.invited_trips) for type-safe call sites.

The cached value of the relation field on source is invalidated after the write so the next access re-fetches fresh data from the graph.

Parameters:

source (Any)
field_name (str | FieldDescriptor)
target (Any)

Return type:

None

async execute(cypher, params=None, write=False)[source]¶

Execute raw Cypher; returns QueryResult; no entity mapping.

Parameters:

cypher (str)
params (dict[str, Any] | None)
write (bool)

Return type:

Any

query(cls)[source]¶

Return an AsyncQueryBuilder for cls.

Async entry point for the fluent query builder. Use await on the terminal methods (all(), one(), count(), etc.):

async with AsyncSession(graph) as session:
    users = await (
        session.query(User).where(User.active == True).limit(20).all()
    )

Parameters:: cls (type[Any])
Return type:: Any

fulltext_search(cls, *, query, fields=None)[source]¶

Async fulltext search; mirrors fulltext_search().

Parameters:

cls (type[Any])
query (str)
fields (list[str] | None)

Return type:

Any

vector_search(cls, *, field, vector, k=10)[source]¶

Async vector KNN search; mirrors vector_search().

Parameters:

cls (type[Any])
field (Any)
vector (list[float])
k (int)

Return type:

Any

async close()[source]¶

expunge_all() and release the graph connection.

Return type:: None

class runic.orm.session.connection_pool.ConnectionManager(db, graph_name)[source]¶

Bases: object

Manages a FalkorDB connection for use with Session.

Holds the db client and graph name; acquire() returns a graph handle. Full connection pooling can be added in a later phase without changing the API.

Parameters:

db (Any)
graph_name (str)

acquire()[source]¶

Return a graph handle for the configured graph name.

Return type:: Any

release(graph)[source]¶

Release a graph handle back to the pool (no-op in current impl).

Parameters:: graph (Any)
Return type:: None

property graph_name: str¶: The configured graph name.

class runic.orm.session.connection_pool.AsyncConnectionManager(db, graph_name)[source]¶

Bases: object

Async variant of ConnectionManager for AsyncFalkorDB clients.

Parameters:

db (Any)
graph_name (str)

acquire()[source]¶

Return an async graph handle for the configured graph name.

Return type:: Any

async release(graph)[source]¶

Release an async graph handle (no-op in current impl).

Parameters:: graph (Any)
Return type:: None

property graph_name: str¶: The configured graph name.

runic.orm.repository — Repository & Pagination¶

class runic.orm.repository.repository.Repository(session, entity_class)[source]¶

Bases: RepositoryProtocol, Generic

Typed reads and explicit Cypher helpers for one entity type.

Mutations (add, delete) and single-PK lookup (get) belong to the Session. All reads here register returned entities in the session identity map.

Example:

with Session(graph) as session:
    repo = Repository(session, Person)
    all_people = repo.find_all()
    page = repo.find_all_paginated(Pageable(page=0, size=25))

Parameters:

session (Session)
entity_class (type[T])

find_all(fetch=None)[source]¶

Return all entities of this type, with optional eager relationship loading.

Parameters:: fetch (list[str] | None)
Return type:: list[T]

find_all_by_ids(pks, fetch=None)[source]¶

Return entities whose primary keys are in pks.

Parameters:

pks (list[Any])
fetch (list[str] | None)

Return type:

list[T]

count()[source]¶

Return the total number of entities of this type.

Return type:: int

exists(pk)[source]¶

Return True if an entity with pk exists in the graph.

Parameters:: pk (Any)
Return type:: bool

find_all_paginated(pageable)[source]¶

Return a single Page of results for pageable.

Parameters:: pageable (Pageable)
Return type:: Page[T]

query()[source]¶

Return a QueryBuilder for this repository’s entity type.

Shorthand for session.query(self._cls):

repo = Repository(session, User)

# These are equivalent:
users = repo.query().where(User.active == True).all()
users = session.query(User).where(User.active == True).all()

Return type:: QueryBuilder[T]

cypher(query, params=None, *, returns=None, write=False)[source]¶

Execute query and return a typed list.

returns controls decoding: int, str, bool → scalar; dict → column-keyed dicts; any Node subclass → decoded entities registered in the session identity map; None → empty list.

Parameters:

query (str)
params (dict[str, Any] | None)
returns (type | None)
write (bool)

Return type:

list[Any]

cypher_one(query, params=None, *, returns=None, write=False)[source]¶

Execute query and return the first mapped value, or None.

Parameters:

query (str)
params (dict[str, Any] | None)
returns (type | None)
write (bool)

Return type:

Any | None

cypher_raw(query, params=None, *, write=False)[source]¶

Execute query and return the raw QueryResult without entity mapping.

Parameters:

query (str)
params (dict[str, Any] | None)
write (bool)

Return type:

Any

class runic.orm.repository.async_repository.AsyncRepository(session, entity_class)[source]¶

Bases: AsyncRepositoryProtocol, Generic

Async typed reads and explicit Cypher helpers for one entity type.

Mirrors Repository with async methods.

Example:

async with AsyncSession(graph) as session:
    repo = AsyncRepository(session, Trip)
    trips = await repo.find_all()

Parameters:

session (AsyncSession)
entity_class (type[T])

async find_all(fetch=None)[source]¶

Return all entities of this type, with optional eager relationship loading.

Parameters:: fetch (list[str] | None)
Return type:: list[T]

async find_all_by_ids(pks, fetch=None)[source]¶

Return entities whose primary keys are in pks.

Parameters:

pks (list[Any])
fetch (list[str] | None)

Return type:

list[T]

async count()[source]¶

Return the total number of entities of this type.

Return type:: int

async exists(pk)[source]¶

Return True if an entity with pk exists in the graph.

Parameters:: pk (Any)
Return type:: bool

async find_all_paginated(pageable)[source]¶

Return a single Page of results for pageable.

Parameters:: pageable (Pageable)
Return type:: Page[T]

query()[source]¶

Return an AsyncQueryBuilder for this repository’s entity type.

Async counterpart of query(). Use await on the terminal methods:

repo = AsyncRepository(session, User)
users = await repo.query().where(User.active == True).all()

Return type:: Any

async cypher(query, params=None, *, returns=None, write=False)[source]¶

Execute query and return a typed list.

Parameters:

query (str)
params (dict[str, Any] | None)
returns (type | None)
write (bool)

Return type:

list[Any]

async cypher_one(query, params=None, *, returns=None, write=False)[source]¶

Execute query and return the first mapped value, or None.

Parameters:

query (str)
params (dict[str, Any] | None)
returns (type | None)
write (bool)

Return type:

Any | None

async cypher_raw(query, params=None, *, write=False)[source]¶

Execute query and return the raw QueryResult.

Parameters:

query (str)
params (dict[str, Any] | None)
write (bool)

Return type:

Any

class runic.orm.repository.pagination.Pageable(page=0, size=20, sort_by=None, direction='ASC')[source]¶

Bases: object

Describes a page request: zero-based page index, page size, and optional sort.

Example:

pageable = Pageable(page=0, size=25, sort_by="name", direction="ASC")
next_page = pageable.next()

Parameters:

page (int)
size (int)
sort_by (str | None)
direction (str)

property offset: int¶: Zero-based item offset for SKIP in Cypher.

next()[source]¶

Return a Pageable for the next page.

Return type:: Pageable

previous()[source]¶

Return a Pageable for the previous page (clamped to 0).

Return type:: Pageable

first()[source]¶

Return a Pageable for the first page.

Return type:: Pageable

class runic.orm.repository.pagination.Page(items, page_number, size, total_elements)[source]¶

Bases: Generic

A single page of results from a paginated query.

Example:

for entity in page:
    print(entity.name)
print(f"Page {page.page_number} of {page.total_pages}")

Parameters:

items (list[T])
page_number (int)
size (int)
total_elements (int)

property total_pages: int¶: Total number of pages given total_elements and size.

has_next()[source]¶

True if there is at least one more page after this one.

Return type:: bool

has_previous()[source]¶

True if this is not the first page.

Return type:: bool

runic.orm.schema — Index & Schema Management¶

class runic.orm.schema.index_manager.IndexSpec(label, property, index_type)[source]¶

Bases: object

Normalized description of a single declared or existing index/constraint.

Used for comparing declared entity schemas against the live graph state. index_type is one of "RANGE", "FULLTEXT", "VECTOR", or "UNIQUE".

Parameters:

label (str)
property (str)
index_type (str)

runic.orm.schema.index_manager.extract_declared_specs(entity_class)[source]¶

Return IndexSpecs declared via Field descriptors on entity_class.

Rules: - unique=True → UNIQUE constraint (backing RANGE is auto-created by FalkorDB). - index=True (without unique) → RANGE index. - index_type="FULLTEXT" → FULLTEXT index. - index_type="VECTOR" → VECTOR index. - Relationship fields are skipped. - A field with both unique=True and index=True emits only UNIQUE.

Parameters:: entity_class (type)
Return type:: set[IndexSpec]

runic.orm.schema.index_manager.parse_existing_specs(graph)[source]¶

Parse live graph state and return all existing NODE index/constraint specs.

Unique constraints are sourced from list_constraints(). Regular indexes (RANGE / FULLTEXT / VECTOR) are sourced from list_indices(). The RANGE index that FalkorDB auto-creates as backing storage for a UNIQUE constraint is excluded so it is not reported as an extra index during schema diffing.

Parameters:: graph (Any)
Return type:: set[IndexSpec]

class runic.orm.schema.index_manager.IndexManager(graph, meta=None)[source]¶

Bases: object

Creates and manages FalkorDB indexes and constraints from Field declarations.

Binds to a FalkorDB graph handle, not a Session.

Example:

manager = IndexManager(graph)
manager.create_indexes(Person, if_not_exists=True)
manager.ensure_indexes(Trip)

Parameters:

graph (Any)
meta (MetaData | None)

create_indexes(entity_class, *, if_not_exists=True)[source]¶

Create all indexes and constraints declared on entity_class.

When if_not_exists is True (default), specs already present in the live graph are skipped silently. Pass if_not_exists=False to attempt creation unconditionally (the graph will raise on duplicates).

Parameters:

entity_class (type)
if_not_exists (bool)

Return type:

None

ensure_indexes(entity_class)[source]¶

Create missing indexes for entity_class; skip those that already exist.

Parameters:: entity_class (type)
Return type:: None

create_spec(spec)[source]¶

Issue the appropriate FalkorDB API call to create a single IndexSpec.

Parameters:: spec (IndexSpec)
Return type:: None

drop_spec(spec)[source]¶

Issue the appropriate FalkorDB API call to drop a single IndexSpec.

Parameters:: spec (IndexSpec)
Return type:: None

class runic.orm.schema.schema_manager.ValidationResult(is_valid, missing_indexes=<factory>, extra_indexes=<factory>, errors=<factory>)[source]¶

Bases: object

Result of a SchemaManager.validate_schema() call.

Parameters:

is_valid (bool)
missing_indexes (list[IndexSpec])
extra_indexes (list[IndexSpec])
errors (list[str])

is_valid¶

True when declared and existing indexes match exactly.

Type:: bool

missing_indexes¶

Declared but not yet created in the live graph.

Type:: list[runic.orm.schema.index_manager.IndexSpec]

extra_indexes¶

Present in the graph but not declared on any entity.

Type:: list[runic.orm.schema.index_manager.IndexSpec]

errors¶

Non-fatal messages collected during validation (e.g., introspection failures).

Type:: list[str]

class runic.orm.schema.schema_manager.SchemaManager(graph, meta=None)[source]¶

Bases: object

Validates and synchronizes FalkorDB indexes against entity Field declarations.

Binds to a FalkorDB graph handle, not a Session.

Example:

schema = SchemaManager(graph)
result = schema.validate_schema([Person, Trip, Stop])

if not result.is_valid:
    print(schema.get_schema_diff([Person, Trip, Stop]))
    schema.sync_schema([Person, Trip, Stop], drop_extra=False)

Parameters:

graph (Any)
meta (MetaData | None)

validate_schema(entity_classes)[source]¶

Compare declared indexes against the live graph state.

Returns a ValidationResult describing missing and extra indexes. is_valid is True only when both sets are empty and no errors occurred.

Parameters:: entity_classes (list[type])
Return type:: ValidationResult

sync_schema(entity_classes, *, drop_extra=False)[source]¶

Create missing indexes and, when drop_extra is True, remove extra ones.

Runs validate_schema internally; no duplicate graph introspection. Extra indexes are only dropped when explicitly requested to prevent data loss.

Parameters:

entity_classes (list[type])
drop_extra (bool)

Return type:

None

get_schema_diff(entity_classes)[source]¶

Return a human-readable diff of declared vs existing indexes.

Lines are prefixed with MISSING or EXTRA; returns a single “in sync” message when no differences exist.

Parameters:: entity_classes (list[type])
Return type:: str

get_schema_info(entity_classes)[source]¶

Return diagnostic information about the current schema state.

The returned dict contains counts of declared/existing indexes plus the full missing and extra lists for programmatic inspection.

Parameters:: entity_classes (list[type])
Return type:: dict[str, Any]

runic.orm.exceptions¶

exception runic.orm.exceptions.OrmError[source]¶

Bases: Exception

Base exception for runic ORM errors.

exception runic.orm.exceptions.EntityNotFoundError[source]¶

Bases: OrmError

Raised when an entity cannot be found by its primary key.

exception runic.orm.exceptions.DetachedEntityError[source]¶

Bases: OrmError

Raised when an operation is attempted on a detached entity.

exception runic.orm.exceptions.LazyLoadError[source]¶

Bases: OrmError

Raised when lazy relationship loading cannot be performed.

Occurs when accessing a lazy field on an entity in an async session (where __get__ cannot await) or when the session is unavailable. Use fetch=[field_name] on session.get() for eager loading instead.

exception runic.orm.exceptions.FieldValidationError[source]¶

Bases: OrmError

Raised when a field value fails validation.

exception runic.orm.exceptions.MetadataError[source]¶

Bases: OrmError

Raised for metadata registry errors (e.g. duplicate labels).

runic.orm.query¶

class runic.orm.query.builder.QueryBuilder(session, root_cls)[source]¶

Bases: Generic[T]

Fluent Cypher query builder for a single root Node class.

Construct via Session.query():

q = session.query(User)

All non-terminal methods return self so calls can be chained:

users = session.query(User).where(User.active == True).limit(10).all()

Parameters:

session (Any) – The Session (or AsyncSession) this builder is bound to.
root_cls (type[T]) – The root Node subclass to query.

alias(name)[source]¶

Set the Cypher variable for the root (most recent) node.

Call immediately after Session.query() to name the root variable, or after TraversalStep.alias() has already been called to rename the last registered target.

Example:

session.query(User).alias("u").where(User.active == True, on="u")

Parameters:: name (str)
Return type:: QueryBuilder[T]

where(expr, *, on=None)[source]¶

Add a WHERE predicate.

Parameters:

expr (Expr) –
A FilterExpr, CompoundExpr, or NegatedExpr. Created via field descriptor operators:
```
User.name == "Alice"
(User.age > 18) & (User.active == True)
```
on (str | None) –
Override the Cypher variable for this predicate. Useful when the same Node class appears under multiple aliases, or when filtering on edge properties:
```
.where(Rated.score > 4.0, on="r")
```

Return type:

QueryBuilder[T]

Notes

Multiple .where() calls are combined with AND. To express OR, use the | operator on the expressions before passing:

.where((User.role == "admin") | (User.role == "mod"))

traverse(relation_field, *, edge_alias=None, optional=True)[source]¶

Traverse a declared Relation() field.

Returns a TraversalStep; call .alias("f") on it to name the target node and return to the builder.

Parameters:

relation_field (FieldDescriptor) –

The Relation-backed field descriptor accessed at class level:

User.friends  # list[User] = Relation(...)
User.rated  # list[Movie] = Relation(edge_model=Rated)

edge_alias (str | None) –
When given, a named relationship variable is emitted in the pattern:
```
(u)-[r:RATED]->(m)
```
This enables filtering on edge properties via .where(Rated.score > 4, on="r") and retrieving edge instances via .all_with_edges().
optional (bool) – True (default) → OPTIONAL MATCH (left-join; keeps source nodes that have no such relationship). False → MATCH (inner join; drops source nodes without a matching relationship).

Returns:

Call .alias("name") on the return value to complete the step.

Return type:

TraversalStep

Examples

# Basic traversal
q = session.query(User).alias("u")
q = q.traverse(User.friends).alias("f")

# Traversal with edge properties
q = session.query(User).alias("u")
q = q.traverse(User.rated, edge_alias="r").alias("m")
q = q.where(Rated.score >= 4.0, on="r")

repeat(relation_field, *, min_hops=1, max_hops=None, optional=False)[source]¶

Traverse a relation with variable-length path quantifier *min..max.

Generates a Cypher pattern like:

(p)-[:PARENT*1..5]->(ancestor:Person)

Parameters:

relation_field (FieldDescriptor) – The Relation field to traverse repeatedly.
min_hops (int) – Minimum number of hops (default 1).
max_hops (int | None) – Maximum number of hops. None means unbounded (*min..).
optional (bool) – False (default for repeat) — required traversal. True → OPTIONAL MATCH.

Returns:

Call .alias("name") to complete the step.

Return type:

TraversalStep

Examples

# All ancestors up to depth 5
ancestors = (
    session.query(Person)
    .alias("p")
    .where(Person.id == start_id)
    .repeat(Person.parent, min_hops=1, max_hops=5)
    .alias("anc")
    .all()
)

# All reachable nodes (unbounded)
reachable = (
    session.query(Node)
    .alias("s")
    .repeat(Node.connected_to)
    .alias("t")
    .all()
)

with_(*aliases)[source]¶

Insert a WITH clause to pipeline results between query stages.

Use when you want to filter/aggregate in one stage before continuing a traversal in the next:

(
    session.query(User)
    .alias("u")
    .where(User.active == True)
    .with_("u")  # WITH u
    .traverse(User.posts)
    .alias("p")
    .return_target("p")
    .all()
)

Parameters:: *aliases (str) – Cypher variable names to carry forward (e.g. "u", "f").
Return type:: QueryBuilder[T]

order_by(field, *, desc=False)[source]¶

Add an ORDER BY term.

Parameters:

field (FieldDescriptor | str) – A field descriptor (User.name) or a raw Cypher expression string ("n.created_at DESC").
desc (bool) – True for descending order (default False).

Return type:

QueryBuilder[T]

Examples

q.order_by(User.age)  # ORDER BY n.age ASC
q.order_by(User.created_at, desc=True)  # ORDER BY n.created_at DESC
q.order_by("score ASC")  # raw string

limit(n)[source]¶

Set LIMIT n on the query.

Parameters:: n (int)
Return type:: QueryBuilder[T]

skip(n)[source]¶

Set SKIP n (offset) on the query.

Parameters:: n (int)
Return type:: QueryBuilder[T]

distinct()[source]¶

Add DISTINCT to the RETURN clause.

Return type:: QueryBuilder[T]

return_target(alias)[source]¶

Set the single alias to return decoded Node instances from.

When a traversal is involved, this selects which alias’s nodes constitute the result of .all():

q.return_target("f")  # returns f-nodes as list[FriendType]

Parameters:: alias (str)
Return type:: QueryBuilder[T]

return_nodes(*aliases)[source]¶

Declare multiple node aliases to include in the RETURN clause.

Used with return_edge() and all_with_edges() to return structured tuples:

q.return_nodes("u", "m").return_edge("r").all_with_edges()

Parameters:: aliases (str)
Return type:: QueryBuilder[T]

return_edge(alias)[source]¶

Declare an edge alias to include in the RETURN clause.

Requires that the traversal was created with edge_alias=alias. The edge is decoded via decode_edge() and included as the middle element of tuples returned by all_with_edges().

Parameters:: alias (str)
Return type:: QueryBuilder[T]

project(*fields)[source]¶

Return only specific property values (scalar projection).

Terminal method .scalars() returns the projected values as a flat list; .all_rows() returns a list of dicts:

# Scalar list
names = session.query(User).project(User.name).scalars()

# Dict list
rows = session.query(User).project(User.name, User.age).all_rows()

Parameters:: fields (FieldDescriptor | str)
Return type:: QueryBuilder[T]

aggregate(*agg_exprs, group_by=None)[source]¶

Add aggregation expressions to the RETURN clause.

Parameters:

*agg_exprs (AggExpr) – One or more AggExpr instances created by the helper functions count(), avg(), etc.
group_by (str | None) –
Alias to keep in the RETURN clause alongside the aggregations (Cypher grouping is implicit — any non-aggregated return term acts as a GROUP BY key):
```
.aggregate(count("*").as_("friend_count"), group_by="u")
# RETURN u, count(*) AS friend_count
```

Return type:

QueryBuilder[T]

Examples

from runic.orm.query import count, avg

result = (
    session.query(User)
    .alias("u")
    .traverse(User.friends)
    .aggregate(count("*").as_("friend_count"), group_by="u")
    .all_rows()  # list[dict] with {"u": ..., "friend_count": int}
)

avg_age = (
    session.query(User).aggregate(avg(User.age).as_("average_age")).scalar()
)

build()[source]¶

Compile the accumulated builder state to a (cypher, params) pair.

This is the core compilation step; all terminal methods call it internally. You can also call it directly for debugging or to integrate with custom execution logic:

cypher, params = session.query(User).where(User.active == True).build()
print(cypher)
# MATCH (n:User)
# WHERE n.active = $p0
# RETURN n

Returns:: A (cypher_string, params_dict) pair ready to pass to execute().
Return type:: tuple[str, dict[str, Any]]

all()[source]¶

Execute and return all matching Node instances.

The return type is the root class (or the alias set by return_target()). Results are decoded and registered in the session identity map.

Returns:: Decoded Node instances of the root type (or target type when return_target() was called).
Return type:: list[T]

one()[source]¶

Execute and return the first matching Node instance, or None.

Internally calls .limit(1).all() and returns the first element.

Return type:: T | None

all_with_edges()[source]¶

Execute and return tuples of (NodeA, EdgeModel, NodeB).

Requires return_nodes() to specify node aliases and return_edge() to specify the edge alias. The edge is decoded via decode_edge().

Returns:: Each element is a tuple whose order matches the aliases given to return_nodes() with the edge inserted at its position in return_edge().
Return type:: list[tuple]

Example

rows = (
    session.query(User)
    .alias("u")
    .traverse(User.rated, edge_alias="r")
    .alias("m")
    .return_nodes("u", "m")
    .return_edge("r")
    .all_with_edges()
)
for user, rated_edge, movie in rows:
    print(f"{user.name} rated {movie.title} with {rated_edge.score}")

all_rows()[source]¶

Execute and return raw column-keyed dicts.

Useful for multi-alias returns, aggregations, or scalar projections where mixed types are in the result set:

rows = q.aggregate(count("*").as_("n"), group_by="u").all_rows()
# [{"u": <User>, "n": 5}, ...]

Return type:: list[dict[str, Any]]

count()[source]¶

Execute a count(*) variant and return the integer count.

Overrides any existing RETURN spec to emit RETURN count(*). Ignores limit() and skip().

Return type:: int

scalar()[source]¶

Execute and return the first column of the first row, or None.

Return type:: Any

scalars()[source]¶

Execute and return the first column of every row as a flat list.

Return type:: list[Any]

register_traversal(fd, source_alias, target_alias, *, optional, edge_alias, min_hops, max_hops)[source]¶

Append a MATCH clause for one traversal step and register aliases.

Called by TraversalStep.alias() to complete a traversal step.

Parameters:

fd (FieldDescriptor)
source_alias (str)
target_alias (str)
optional (bool)
edge_alias (str | None)
min_hops (int)
max_hops (int | None)

Return type:

QueryBuilder[T]

class runic.orm.query.builder.AsyncQueryBuilder(session, root_cls)[source]¶

Bases: QueryBuilder[T]

Async variant of QueryBuilder for use with AsyncSession.

All intermediate (chainable) methods are identical to the sync version. Only the terminal methods are replaced with async def equivalents.

Example

async with AsyncSession(graph) as session:
    users = await (
        session.query(User)
        .where(User.active == True)
        .order_by(User.name)
        .limit(50)
        .all()
    )

Parameters:

session (Any)
root_cls (type[T])

async all()[source]¶

Async version of all().

Return type:: list[T]

async one()[source]¶

Async version of one().

Return type:: T | None

async all_with_edges()[source]¶

Async version of all_with_edges().

Return type:: list[tuple[Any, …]]

async all_rows()[source]¶

Async version of all_rows().

Return type:: list[dict[str, Any]]

async count()[source]¶

Async version of count().

Return type:: int

async scalar()[source]¶

Async version of scalar().

Return type:: Any

async scalars()[source]¶

Async version of scalars().

Return type:: list[Any]

class runic.orm.query.builder.FulltextQueryBuilder(session, root_cls, query, fields=None)[source]¶

Bases: QueryBuilder[T]

QueryBuilder variant for FalkorDB fulltext search queries.

Constructed via fulltext_search(). The root MATCH is replaced with a CALL db.idx.fulltext.queryNodes(...) invocation that uses the declared fulltext index.

The fulltext index must have been created for the node’s label, e.g.:

class Post(Node, labels=["Post"]):
    title: str = Field(index_type="FULLTEXT")

Example

posts = (
    session.fulltext_search(Post, query="graph databases", fields=["title"])
    .where(Post.published == True)
    .order_by(Post.created_at, desc=True)
    .limit(20)
    .all()
)

Cypher emitted:

CALL db.idx.fulltext.queryNodes('Post', $__fts_query) YIELD node AS n
WHERE n.published = $p0
RETURN n
ORDER BY n.created_at DESC
LIMIT 20

Parameters:

session (Any)
root_cls (type[T])
query (str)
fields (list[str] | None)

build()[source]¶

Compile to Cypher, replacing MATCH with CALL fulltext procedure.

Return type:: tuple[str, dict[str, Any]]

class runic.orm.query.builder.VectorQueryBuilder(session, root_cls, field, vector, k)[source]¶

Bases: QueryBuilder[T]

QueryBuilder variant for FalkorDB vector KNN search.

Constructed via vector_search(). Appends a KNN distance expression to the ORDER BY and RETURN clauses.

The field must have index_type="VECTOR" and an HNSW vector index must be created via SchemaManager():

class Document(Node, labels=["Document"]):
    embedding: Vector = Field(index_type="VECTOR")

Example

similar = (
    session.vector_search(
        Document,
        field=Document.embedding,
        vector=[0.1, 0.2, 0.3],
        k=10,
    )
    .where(Document.active == True)
    .all()
)

Cypher emitted (FalkorDB KNN syntax):

MATCH (n:Document)
WHERE n.active = $p0
RETURN n, vecf32(n.embedding) <-> vecf32($__knn_vec) AS __score
ORDER BY __score ASC
LIMIT 10

Note

The exact FalkorDB KNN Cypher syntax may vary by version. If the above pattern does not work, use repo.cypher() with a hand-written query.

Parameters:

session (Any)
root_cls (type[T])
field (FieldDescriptor)
vector (list[float])
k (int)

build()[source]¶

Compile to Cypher with KNN ORDER BY.

Return type:: tuple[str, dict[str, Any]]

class runic.orm.query.traversal.TraversalStep(builder, field_descriptor, source_alias, *, optional=True, edge_alias=None, min_hops=1, max_hops=1)[source]¶

Bases: object

Pending traversal hop; returned by QueryBuilder.traverse() and QueryBuilder.repeat().

Call alias() to complete the step and resume the builder chain.

Parameters:

builder (QueryBuilder[Any]) – The owning QueryBuilder instance.
field_descriptor (Any) – The FieldDescriptor for the Relation field being traversed.
source_alias (str) – The Cypher variable name of the source node.
optional (bool) – When True (the default), the traversal emits OPTIONAL MATCH (left-join: source nodes without the relationship are still returned). When False, emits MATCH (inner-join: drops source nodes that have no such relationship).
edge_alias (str | None) – Optional Cypher variable name for the relationship itself. When set, the generated pattern is (src)-[edge_alias:TYPE]->(tgt) instead of the anonymous (src)-[:TYPE]->(tgt), enabling edge property filtering and retrieval.
min_hops (int) – Minimum number of hops for variable-length paths (default 1). Values > 1 only take effect when combined with max_hops to produce a *min..max quantifier.
max_hops (int | None) – Maximum number of hops. None means unbounded (*min..). A value of 1 with min_hops=1 produces a fixed single-hop pattern.

alias(name)[source]¶

Calling this method:

Resolves the target Node class from the Relation field’s target.
Appends the appropriate (OPTIONAL) MATCH clause to the builder.
Registers name → target_cls in the builder’s alias map.
Registers edge_alias → Edge class if an edge alias was given.
Sets the builder’s last alias (used as the default RETURN target when no explicit return_target() is called).

Parameters:: name (str) – Cypher variable name for the target node (e.g. "f", "m").
Returns:: The owning builder, ready for continued chaining.
Return type:: QueryBuilder

class runic.orm.query.expressions.Expr[source]¶

Bases: object

Abstract base for all filter/compound/negated expressions.

Subclass instances are passed to QueryBuilder.where(). They support & (AND), | (OR), and ~ (NOT) to build composite predicates:

q.where((User.age > 18) & (User.active == True))

class runic.orm.query.expressions.FilterExpr(cls, prop, op, value=None, alias=None, negate=False)[source]¶

Bases: Expr

A single WHERE predicate: alias.prop OP $pN.

Created automatically by the FieldDescriptor operator overloads; you do not normally instantiate this class directly.

Parameters:

cls (type)
prop (str)
op (str)
value (Any)
alias (str | None)
negate (bool)

cls¶

The ORM Node or Edge subclass that owns the field. Used by the query builder to look up the Cypher variable (alias) for the class in the current query context.

Type:: type

prop¶

The property name as declared on the ORM class.

Type:: str

op¶

A Cypher operator string: "=", "<>", ">", ">=", "<", "<=", "CONTAINS", "STARTS WITH", "=~" (regex), "IN", "IS NULL", "IS NOT NULL".

Type:: str

value¶

The Python value to compare against. None for null-check ops. TypeConverter.to_graph() is applied during Cypher compilation if the field has a converter; cypher_fn wraps the param reference.

Type:: Any

alias¶

Optional explicit Cypher variable override (set via on= in QueryBuilder.where()). When None, the builder derives the alias from cls.

Type:: str | None

negate¶

When True, wraps the predicate in NOT (...). Used for not_in_(); prefer the ~ operator for general negation.

Type:: bool

with_alias(alias)[source]¶

Return a copy of this expression with alias as the explicit variable.

Parameters:: alias (str)
Return type:: FilterExpr

class runic.orm.query.expressions.CompoundExpr(op, operands=<factory>)[source]¶

Bases: Expr

AND / OR combination of multiple sub-expressions.

Example:

(User.age > 18) & (User.active == True)
# → CompoundExpr(op="AND", operands=[FilterExpr(...), FilterExpr(...)])

Parameters:

op (Literal['AND', 'OR'])
operands (list[Expr])

class runic.orm.query.expressions.NegatedExpr(operand)[source]¶

Bases: Expr

NOT wrapper around another expression.

Example:

~(User.banned == True)
# → NegatedExpr(operand=FilterExpr(...))

Parameters:: operand (Expr)

class runic.orm.query.expressions.OrderExpr(alias, prop, raw=None, desc=False)[source]¶

Bases: object

Represents a single ORDER BY term.

Created by QueryBuilder.order_by(); not usually instantiated directly.

Parameters:

alias (str | None)
prop (str | None)
raw (str | None)
desc (bool)

alias¶

The Cypher variable (e.g. "n", "u").

Type:: str | None

prop¶

The property name (e.g. "age"). None when raw is set.

Type:: str | None

raw¶

A raw Cypher expression string (e.g. "score ASC"). Used when the user passes a string directly to order_by().

Type:: str | None

desc¶

True for descending order; False (default) for ascending.

Type:: bool

to_cypher()[source]¶

Render to a Cypher ORDER BY term string.

Return type:: str

class runic.orm.query.expressions.AggExpr(func, field='*', result_alias=None, distinct=False)[source]¶

Bases: object

An aggregation function expression for use in RETURN clauses.

Created via the helper functions count(), avg(), sum_(), min_(), max_(), collect().

Parameters:

func (str)
field (Any)
result_alias (str | None)
distinct (bool)

func¶

Cypher aggregation function name: "count", "avg", "sum", "min", "max", "collect".

Type:: str

field¶

A FieldDescriptor or raw string ("*" for count(*)).

Type:: Any

result_alias¶

The AS name alias in the RETURN clause.

Type:: str | None

distinct¶

When True, emits count(DISTINCT n.prop) etc.

Type:: bool

as_(alias)[source]¶

Return a copy with a RETURN alias set.

Parameters:: alias (str)
Return type:: AggExpr

to_cypher(alias_map)[source]¶

Render to a Cypher aggregation expression string.

Parameters:: alias_map (dict[type, str]) – Mapping from ORM class to Cypher variable (provided by the builder during compilation).
Return type:: str

runic.orm.query.expressions.count(field='*', *, distinct=False)[source]¶

Create a count(...) aggregation expression.

Parameters:

field (Any) – The field to count, or "*" (default) for count(*).
distinct (bool) – When True, emits count(DISTINCT field).

Return type:

AggExpr

Examples

q.aggregate(count())  # count(*)
q.aggregate(count(User.name, distinct=True))  # count(DISTINCT n.name)

runic.orm.query.expressions.avg(field)[source]¶

Create an avg(...) aggregation expression.

Example:

q.aggregate(avg(User.age).as_("average_age"))

Parameters:: field (Any)
Return type:: AggExpr

runic.orm.query.expressions.sum_(field)[source]¶

Create a sum(...) aggregation expression.

Example:

q.aggregate(sum_(Order.amount).as_("total"))

Parameters:: field (Any)
Return type:: AggExpr

runic.orm.query.expressions.min_(field)[source]¶

Create a min(...) aggregation expression.

Parameters:: field (Any)
Return type:: AggExpr

runic.orm.query.expressions.max_(field)[source]¶

Create a max(...) aggregation expression.

Parameters:: field (Any)
Return type:: AggExpr

runic.orm.query.expressions.collect(field, *, distinct=False)[source]¶

Create a collect(...) aggregation expression.

Collects values from multiple rows into a list.

Example:

q.aggregate(collect(Post.title).as_("post_titles"))

Parameters:

field (Any)
distinct (bool)

Return type:

AggExpr