uni_pydantic

uni-pydantic: Pydantic-based OGM for Uni Graph Database.

This package provides a type-safe Object-Graph Mapping layer on top of the Uni graph database, using Pydantic v2 for model definitions.

Example:

from uni_db import Uni from uni_pydantic import UniNode, UniSession, Field, Relationship, Vector

class Person(UniNode): ... name: str ... age: int | None = None ... email: str = Field(unique=True) ... embedding: Vector[1536] ... friends: list["Person"] = Relationship("FRIEND_OF", direction="both")

db = Uni("./my_graph") session = UniSession(db) session.register(Person) session.sync_schema()

alice = Person(name="Alice", age=30, email="alice@example.com") session.add(alice) session.commit()

Query with type safety

adults = session.query(Person).filter(Person.age >= 18).all()

View Source

  1# SPDX-License-Identifier: Apache-2.0
  2# Copyright 2024-2026 Dragonscale Team
  3
  4"""
  5uni-pydantic: Pydantic-based OGM for Uni Graph Database.
  6
  7This package provides a type-safe Object-Graph Mapping layer on top of
  8the Uni graph database, using Pydantic v2 for model definitions.
  9
 10Example:
 11    >>> from uni_db import Uni
 12    >>> from uni_pydantic import UniNode, UniSession, Field, Relationship, Vector
 13    >>>
 14    >>> class Person(UniNode):
 15    ...     name: str
 16    ...     age: int | None = None
 17    ...     email: str = Field(unique=True)
 18    ...     embedding: Vector[1536]
 19    ...     friends: list["Person"] = Relationship("FRIEND_OF", direction="both")
 20    >>>
 21    >>> db = Uni("./my_graph")
 22    >>> session = UniSession(db)
 23    >>> session.register(Person)
 24    >>> session.sync_schema()
 25    >>>
 26    >>> alice = Person(name="Alice", age=30, email="alice@example.com")
 27    >>> session.add(alice)
 28    >>> session.commit()
 29    >>>
 30    >>> # Query with type safety
 31    >>> adults = session.query(Person).filter(Person.age >= 18).all()
 32"""
 33
 34from pathlib import Path
 35
 36# Base classes
 37# Async support
 38from .async_query import AsyncQueryBuilder
 39from .async_session import AsyncUniSession, AsyncUniTransaction
 40from .base import SearchScores, UniEdge, UniNode
 41
 42# Database wrappers
 43from .database import AsyncUniDatabase, UniDatabase
 44
 45# Exceptions
 46from .exceptions import (
 47    BulkLoadError,
 48    CypherInjectionError,
 49    LazyLoadError,
 50    NotPersisted,
 51    NotRegisteredError,
 52    NotTrackedError,
 53    QueryError,
 54    RelationshipError,
 55    SchemaError,
 56    SessionError,
 57    TransactionError,
 58    TypeMappingError,
 59    UniPydanticError,
 60    ValidationError,
 61)
 62
 63# Field configuration
 64from .fields import (
 65    Direction,
 66    Field,
 67    FieldConfig,
 68    IndexType,
 69    Relationship,
 70    RelationshipConfig,
 71    RelationshipDescriptor,
 72    VectorMetric,
 73    get_field_config,
 74)
 75
 76# Lifecycle hooks
 77from .hooks import (
 78    after_create,
 79    after_delete,
 80    after_load,
 81    after_update,
 82    before_create,
 83    before_delete,
 84    before_load,
 85    before_update,
 86)
 87
 88# Query builder
 89from .query import (
 90    FilterExpr,
 91    FilterOp,
 92    HybridSearchConfig,
 93    ModelProxy,
 94    OrderByClause,
 95    PropertyProxy,
 96    QueryBuilder,
 97    SparseSearchConfig,
 98    TraversalStep,
 99    VectorSearchConfig,
100)
101
102# Schema generation
103from .schema import (
104    DatabaseSchema,
105    EdgeTypeSchema,
106    LabelSchema,
107    PropertySchema,
108    SchemaGenerator,
109    generate_schema,
110)
111
112# Session management
113from .session import UniSession, UniTransaction
114
115# Type utilities
116from .types import (
117    DATETIME_TYPES,
118    Btic,
119    SparseVector,
120    Vector,
121    db_to_python_value,
122    get_sparse_vector_dimensions,
123    get_vector_dimensions,
124    is_list_type,
125    is_optional,
126    python_to_db_value,
127    python_type_to_uni,
128    uni_to_python_type,
129    unwrap_annotated,
130)
131
132
133def _package_version() -> str:
134    """Read the version from installed package metadata.
135
136    Derived rather than hand-written: a literal here is a second source of truth
137    beside ``pyproject.toml`` and drifts silently, because only ``pyproject.toml``
138    is checked at release time. It had drifted to ``2.5.0`` against a ``3.3.0``
139    package, which would have shipped a wrong ``__version__`` to every consumer.
140
141    Falls back to reading ``pyproject.toml`` for a source checkout that has not
142    been installed, and finally to ``"0.0.0"`` so an import never fails over a
143    version string.
144    """
145    from importlib.metadata import PackageNotFoundError, version
146
147    try:
148        return version("uni-pydantic")
149    except PackageNotFoundError:
150        pass
151
152    try:
153        import tomllib
154
155        pyproject = Path(__file__).resolve().parents[2] / "pyproject.toml"
156        with pyproject.open("rb") as fh:
157            return str(tomllib.load(fh)["project"]["version"])
158    except Exception:  # noqa: BLE001 - a version string must never break import
159        return "0.0.0"
160
161
162__version__ = _package_version()
163
164
165__all__ = [
166    # Version
167    "__version__",
168    # Base classes
169    "UniNode",
170    "UniEdge",
171    "SearchScores",
172    # Session
173    "UniSession",
174    "UniTransaction",
175    # Async Session
176    "AsyncUniSession",
177    "AsyncUniTransaction",
178    # Fields
179    "Field",
180    "FieldConfig",
181    "Relationship",
182    "RelationshipConfig",
183    "RelationshipDescriptor",
184    "get_field_config",
185    "IndexType",
186    "Direction",
187    "VectorMetric",
188    # Types
189    "Btic",
190    "Vector",
191    "SparseVector",
192    "python_type_to_uni",
193    "uni_to_python_type",
194    "get_vector_dimensions",
195    "get_sparse_vector_dimensions",
196    "is_optional",
197    "is_list_type",
198    "unwrap_annotated",
199    "python_to_db_value",
200    "db_to_python_value",
201    "DATETIME_TYPES",
202    # Query
203    "QueryBuilder",
204    "AsyncQueryBuilder",
205    "FilterExpr",
206    "FilterOp",
207    "PropertyProxy",
208    "ModelProxy",
209    "OrderByClause",
210    "TraversalStep",
211    "VectorSearchConfig",
212    "SparseSearchConfig",
213    "HybridSearchConfig",
214    # Schema
215    "SchemaGenerator",
216    "DatabaseSchema",
217    "LabelSchema",
218    "EdgeTypeSchema",
219    "PropertySchema",
220    "generate_schema",
221    # Database
222    "UniDatabase",
223    "AsyncUniDatabase",
224    # Hooks
225    "before_create",
226    "after_create",
227    "before_update",
228    "after_update",
229    "before_delete",
230    "after_delete",
231    "before_load",
232    "after_load",
233    # Exceptions
234    "UniPydanticError",
235    "SchemaError",
236    "TypeMappingError",
237    "ValidationError",
238    "SessionError",
239    "NotRegisteredError",
240    "NotPersisted",
241    "NotTrackedError",
242    "TransactionError",
243    "QueryError",
244    "RelationshipError",
245    "LazyLoadError",
246    "BulkLoadError",
247    "CypherInjectionError",
248]

__version__ = '3.4.0'

@dataclass(frozen=True)

class SearchScores: View Source

191@dataclass(frozen=True)
192class SearchScores:
193    """Relevance scores attached to a hydrated search result.
194
195    Populated on instances returned by the search builders
196    (``vector_search`` / ``sparse_search`` / ``hybrid_search``) and reachable
197    via :attr:`UniNode.search_scores`. Kept in a sidecar (rather than as a model
198    field) so it never collides with a user field named ``score`` and survives
199    hydration under ``extra="forbid"``.
200
201    ``score`` is the primary score used for ordering (the fused score for
202    hybrid; the per-source score for single-source search). The per-arm fields
203    are populated only when that retrieval source contributed, ``None`` otherwise.
204    """
205
206    score: float
207    vector: float | None = None
208    fts: float | None = None
209    sparse: float | None = None
210    rerank: float | None = None
211    distance: float | None = None

Relevance scores attached to a hydrated search result.

Populated on instances returned by the search builders (vector_search / sparse_search / hybrid_search) and reachable via UniNode.search_scores. Kept in a sidecar (rather than as a model field) so it never collides with a user field named score and survives hydration under extra="forbid".

score is the primary score used for ordering (the fused score for hybrid; the per-source score for single-source search). The per-arm fields are populated only when that retrieval source contributed, None otherwise.

score: float

vector: float | None = None

fts: float | None = None

sparse: float | None = None

rerank: float | None = None

distance: float | None = None

class UniTransaction: View Source

 64class UniTransaction:
 65    """
 66    Transaction context for atomic operations.
 67
 68    Provides commit/rollback semantics for a group of operations.
 69
 70    Example:
 71        >>> with session.transaction() as tx:
 72        ...     alice = Person(name="Alice")
 73        ...     tx.add(alice)
 74        ...     # Auto-commits on success, rolls back on exception
 75    """
 76
 77    def __init__(self, session: UniSession) -> None:
 78        self._session = session
 79        self._tx: uni_db.Transaction | None = None
 80        self._pending_nodes: list[UniNode] = []
 81        self._pending_edges: list[tuple[UniNode, str, UniNode, UniEdge | None]] = []
 82        self._committed = False
 83        self._rolled_back = False
 84
 85    def __enter__(self) -> UniTransaction:
 86        self._tx = self._session._db_session.tx()
 87        return self
 88
 89    def __exit__(
 90        self,
 91        exc_type: type[BaseException] | None,
 92        exc_val: BaseException | None,
 93        exc_tb: TracebackType | None,
 94    ) -> None:
 95        if exc_type is not None:
 96            self.rollback()
 97            return
 98        if not self._committed and not self._rolled_back:
 99            self.commit()
100
101    def add(self, entity: UniNode) -> None:
102        """Add a node to be created in this transaction."""
103        self._pending_nodes.append(entity)
104
105    def create_edge(
106        self,
107        source: UniNode,
108        edge_type: str,
109        target: UniNode,
110        properties: UniEdge | None = None,
111        **kwargs: Any,
112    ) -> None:
113        """Create an edge between two nodes in this transaction."""
114        if not source.is_persisted:
115            raise NotPersisted(source)
116        if not target.is_persisted:
117            raise NotPersisted(target)
118        self._pending_edges.append((source, edge_type, target, properties))
119
120    def commit(self) -> None:
121        """Commit the transaction."""
122        if self._committed:
123            raise TransactionError("Transaction already committed")
124        if self._rolled_back:
125            raise TransactionError("Transaction already rolled back")
126
127        if self._tx is None:
128            raise TransactionError("Transaction not started")
129
130        try:
131            # Create pending nodes
132            for node in self._pending_nodes:
133                self._session._create_node_in_tx(node, self._tx)
134
135            # Create pending edges
136            for source, edge_type, target, props in self._pending_edges:
137                self._session._create_edge_in_tx(
138                    source, edge_type, target, props, self._tx
139                )
140
141            self._tx.commit()
142            self._committed = True
143
144            # Mark nodes as clean
145            for node in self._pending_nodes:
146                node._mark_clean()
147
148        except Exception as e:
149            self.rollback()
150            raise TransactionError(f"Commit failed: {e}") from e
151
152    def rollback(self) -> None:
153        """Rollback the transaction."""
154        if self._rolled_back:
155            return
156        if self._tx is not None:
157            self._tx.rollback()
158        self._rolled_back = True
159        self._pending_nodes.clear()
160        self._pending_edges.clear()

Transaction context for atomic operations.

Provides commit/rollback semantics for a group of operations.

Example:

with session.transaction() as tx: ... alice = Person(name="Alice") ... tx.add(alice) ... # Auto-commits on success, rolls back on exception

UniTransaction(session: UniSession) View Source

77    def __init__(self, session: UniSession) -> None:
78        self._session = session
79        self._tx: uni_db.Transaction | None = None
80        self._pending_nodes: list[UniNode] = []
81        self._pending_edges: list[tuple[UniNode, str, UniNode, UniEdge | None]] = []
82        self._committed = False
83        self._rolled_back = False

def add(self, entity: UniNode) -> None: View Source

101    def add(self, entity: UniNode) -> None:
102        """Add a node to be created in this transaction."""
103        self._pending_nodes.append(entity)

Add a node to be created in this transaction.

def create_edge( self, source: UniNode, edge_type: str, target: UniNode, properties: UniEdge | None = None, **kwargs: Any) -> None: View Source

105    def create_edge(
106        self,
107        source: UniNode,
108        edge_type: str,
109        target: UniNode,
110        properties: UniEdge | None = None,
111        **kwargs: Any,
112    ) -> None:
113        """Create an edge between two nodes in this transaction."""
114        if not source.is_persisted:
115            raise NotPersisted(source)
116        if not target.is_persisted:
117            raise NotPersisted(target)
118        self._pending_edges.append((source, edge_type, target, properties))

Create an edge between two nodes in this transaction.

def commit(self) -> None: View Source

120    def commit(self) -> None:
121        """Commit the transaction."""
122        if self._committed:
123            raise TransactionError("Transaction already committed")
124        if self._rolled_back:
125            raise TransactionError("Transaction already rolled back")
126
127        if self._tx is None:
128            raise TransactionError("Transaction not started")
129
130        try:
131            # Create pending nodes
132            for node in self._pending_nodes:
133                self._session._create_node_in_tx(node, self._tx)
134
135            # Create pending edges
136            for source, edge_type, target, props in self._pending_edges:
137                self._session._create_edge_in_tx(
138                    source, edge_type, target, props, self._tx
139                )
140
141            self._tx.commit()
142            self._committed = True
143
144            # Mark nodes as clean
145            for node in self._pending_nodes:
146                node._mark_clean()
147
148        except Exception as e:
149            self.rollback()
150            raise TransactionError(f"Commit failed: {e}") from e

Commit the transaction.

def rollback(self) -> None: View Source

152    def rollback(self) -> None:
153        """Rollback the transaction."""
154        if self._rolled_back:
155            return
156        if self._tx is not None:
157            self._tx.rollback()
158        self._rolled_back = True
159        self._pending_nodes.clear()
160        self._pending_edges.clear()

Rollback the transaction.

class AsyncUniTransaction: View Source

 61class AsyncUniTransaction:
 62    """Async transaction context for atomic operations."""
 63
 64    def __init__(self, session: AsyncUniSession) -> None:
 65        self._session = session
 66        self._tx: uni_db.AsyncTransaction | None = None
 67        self._pending_nodes: list[UniNode] = []
 68        self._pending_edges: list[tuple[UniNode, str, UniNode, UniEdge | None]] = []
 69        self._committed = False
 70        self._rolled_back = False
 71
 72    async def __aenter__(self) -> AsyncUniTransaction:
 73        self._tx = await self._session._db_session.tx()
 74        return self
 75
 76    async def __aexit__(
 77        self,
 78        exc_type: type[BaseException] | None,
 79        exc_val: BaseException | None,
 80        exc_tb: TracebackType | None,
 81    ) -> None:
 82        if exc_type is not None:
 83            await self.rollback()
 84            return
 85        if not self._committed and not self._rolled_back:
 86            await self.commit()
 87
 88    def add(self, entity: UniNode) -> None:
 89        """Add a node to be created in this transaction (sync — just collects)."""
 90        self._pending_nodes.append(entity)
 91
 92    def create_edge(
 93        self,
 94        source: UniNode,
 95        edge_type: str,
 96        target: UniNode,
 97        properties: UniEdge | None = None,
 98    ) -> None:
 99        """Create an edge between two nodes in this transaction (sync — just collects)."""
100        if not source.is_persisted:
101            raise NotPersisted(source)
102        if not target.is_persisted:
103            raise NotPersisted(target)
104        self._pending_edges.append((source, edge_type, target, properties))
105
106    async def commit(self) -> None:
107        """Commit the transaction."""
108        if self._committed:
109            raise TransactionError("Transaction already committed")
110        if self._rolled_back:
111            raise TransactionError("Transaction already rolled back")
112        if self._tx is None:
113            raise TransactionError("Transaction not started")
114
115        try:
116            for node in self._pending_nodes:
117                await self._session._create_node_in_tx(node, self._tx)
118            for source, edge_type, target, props in self._pending_edges:
119                await self._session._create_edge_in_tx(
120                    source, edge_type, target, props, self._tx
121                )
122            await self._tx.commit()
123            self._committed = True
124            for node in self._pending_nodes:
125                node._mark_clean()
126        except Exception as e:
127            await self.rollback()
128            raise TransactionError(f"Commit failed: {e}") from e
129
130    async def rollback(self) -> None:
131        """Rollback the transaction."""
132        if self._rolled_back:
133            return
134        if self._tx is not None:
135            await self._tx.rollback()
136        self._rolled_back = True
137        self._pending_nodes.clear()
138        self._pending_edges.clear()

Async transaction context for atomic operations.

AsyncUniTransaction(session: AsyncUniSession) View Source

64    def __init__(self, session: AsyncUniSession) -> None:
65        self._session = session
66        self._tx: uni_db.AsyncTransaction | None = None
67        self._pending_nodes: list[UniNode] = []
68        self._pending_edges: list[tuple[UniNode, str, UniNode, UniEdge | None]] = []
69        self._committed = False
70        self._rolled_back = False

def add(self, entity: UniNode) -> None: View Source

88    def add(self, entity: UniNode) -> None:
89        """Add a node to be created in this transaction (sync — just collects)."""
90        self._pending_nodes.append(entity)

Add a node to be created in this transaction (sync — just collects).

def create_edge( self, source: UniNode, edge_type: str, target: UniNode, properties: UniEdge | None = None) -> None: View Source

 92    def create_edge(
 93        self,
 94        source: UniNode,
 95        edge_type: str,
 96        target: UniNode,
 97        properties: UniEdge | None = None,
 98    ) -> None:
 99        """Create an edge between two nodes in this transaction (sync — just collects)."""
100        if not source.is_persisted:
101            raise NotPersisted(source)
102        if not target.is_persisted:
103            raise NotPersisted(target)
104        self._pending_edges.append((source, edge_type, target, properties))

Create an edge between two nodes in this transaction (sync — just collects).

async def commit(self) -> None: View Source

106    async def commit(self) -> None:
107        """Commit the transaction."""
108        if self._committed:
109            raise TransactionError("Transaction already committed")
110        if self._rolled_back:
111            raise TransactionError("Transaction already rolled back")
112        if self._tx is None:
113            raise TransactionError("Transaction not started")
114
115        try:
116            for node in self._pending_nodes:
117                await self._session._create_node_in_tx(node, self._tx)
118            for source, edge_type, target, props in self._pending_edges:
119                await self._session._create_edge_in_tx(
120                    source, edge_type, target, props, self._tx
121                )
122            await self._tx.commit()
123            self._committed = True
124            for node in self._pending_nodes:
125                node._mark_clean()
126        except Exception as e:
127            await self.rollback()
128            raise TransactionError(f"Commit failed: {e}") from e

Commit the transaction.

async def rollback(self) -> None: View Source

130    async def rollback(self) -> None:
131        """Rollback the transaction."""
132        if self._rolled_back:
133            return
134        if self._tx is not None:
135            await self._tx.rollback()
136        self._rolled_back = True
137        self._pending_nodes.clear()
138        self._pending_edges.clear()

Rollback the transaction.

def Field( default: Any = Ellipsis, *, default_factory: Callable[[], typing.Any] | None = None, alias: str | None = None, title: str | None = None, description: str | None = None, examples: list[typing.Any] | None = None, exclude: bool = False, json_schema_extra: dict[str, typing.Any] | None = None, index: Optional[Literal['btree', 'hash', 'fulltext', 'vector', 'sparse']] = None, unique: bool = False, tokenizer: str | None = None, metric: Optional[Literal['l2', 'cosine', 'dot']] = None, generated: str | None = None) -> Any: View Source

 69def Field(
 70    default: Any = ...,
 71    *,
 72    default_factory: Callable[[], Any] | None = None,
 73    alias: str | None = None,
 74    title: str | None = None,
 75    description: str | None = None,
 76    examples: list[Any] | None = None,
 77    exclude: bool = False,
 78    json_schema_extra: dict[str, Any] | None = None,
 79    # Uni-specific options
 80    index: IndexType | None = None,
 81    unique: bool = False,
 82    tokenizer: str | None = None,
 83    metric: VectorMetric | None = None,
 84    generated: str | None = None,
 85) -> Any:
 86    """
 87    Create a field with uni-pydantic configuration.
 88
 89    This extends Pydantic's Field with graph database options.
 90
 91    Args:
 92        default: Default value for the field.
 93        default_factory: Factory function for default value.
 94        alias: Field alias for serialization.
 95        title: Human-readable title.
 96        description: Field description.
 97        examples: Example values.
 98        exclude: Exclude from serialization.
 99        json_schema_extra: Extra JSON schema properties.
100        index: Index type ("btree", "hash", "fulltext", "vector").
101        unique: Whether to create a unique constraint.
102        tokenizer: Tokenizer for fulltext index (default: "standard").
103        metric: Distance metric for vector index ("l2", "cosine", "dot").
104        generated: Expression for generated/computed property.
105
106    Returns:
107        A Pydantic FieldInfo with uni-pydantic metadata attached.
108
109    Examples:
110        >>> class Person(UniNode):
111        ...     name: str = Field(index="btree")
112        ...     email: str = Field(unique=True)
113        ...     bio: str = Field(index="fulltext", tokenizer="standard")
114        ...     embedding: Vector[768] = Field(metric="cosine")
115    """
116    # Default tokenizer for fulltext indexes
117    if index == "fulltext" and tokenizer is None:
118        tokenizer = "standard"
119
120    # Store uni config in json_schema_extra
121    uni_config = FieldConfig(
122        index=index,
123        unique=unique,
124        tokenizer=tokenizer,
125        metric=metric,
126        generated=generated,
127        default=default,
128        default_factory=default_factory,
129        alias=alias,
130        title=title,
131        description=description,
132        examples=examples,
133        exclude=exclude,
134        json_schema_extra=json_schema_extra,
135    )
136
137    # Merge uni config into json_schema_extra
138    extra = json_schema_extra or {}
139    extra["uni_config"] = uni_config
140
141    # Create Pydantic FieldInfo
142    from pydantic.fields import FieldInfo as PydanticFieldInfo
143
144    if default_factory is not None:
145        return PydanticFieldInfo(
146            default_factory=default_factory,
147            alias=alias,
148            title=title,
149            description=description,
150            examples=examples,
151            exclude=exclude,
152            json_schema_extra=extra,
153        )
154    elif default is not ...:
155        return PydanticFieldInfo(
156            default=default,
157            alias=alias,
158            title=title,
159            description=description,
160            examples=examples,
161            exclude=exclude,
162            json_schema_extra=extra,
163        )
164    else:
165        return PydanticFieldInfo(
166            alias=alias,
167            title=title,
168            description=description,
169            examples=examples,
170            exclude=exclude,
171            json_schema_extra=extra,
172        )

Create a field with uni-pydantic configuration.

This extends Pydantic's Field with graph database options.

Args: default: Default value for the field. default_factory: Factory function for default value. alias: Field alias for serialization. title: Human-readable title. description: Field description. examples: Example values. exclude: Exclude from serialization. json_schema_extra: Extra JSON schema properties. index: Index type ("btree", "hash", "fulltext", "vector"). unique: Whether to create a unique constraint. tokenizer: Tokenizer for fulltext index (default: "standard"). metric: Distance metric for vector index ("l2", "cosine", "dot"). generated: Expression for generated/computed property.

Returns: A Pydantic FieldInfo with uni-pydantic metadata attached.

Examples:

class Person(UniNode): ... name: str = Field(index="btree") ... email: str = Field(unique=True) ... bio: str = Field(index="fulltext", tokenizer="standard") ... embedding: Vector[768] = Field(metric="cosine")

@dataclass

class FieldConfig: View Source

41@dataclass
42class FieldConfig:
43    """Configuration for a uni-pydantic field."""
44
45    # Index configuration
46    index: IndexType | None = None
47    unique: bool = False
48
49    # Fulltext index options
50    tokenizer: str | None = None
51
52    # Vector index options
53    metric: VectorMetric | None = None
54
55    # Generated/computed property
56    generated: str | None = None
57
58    # Pydantic field options (passed through)
59    default: Any = dataclass_field(default_factory=lambda: ...)
60    default_factory: Callable[[], Any] | None = None
61    alias: str | None = None
62    title: str | None = None
63    description: str | None = None
64    examples: list[Any] | None = None
65    exclude: bool = False
66    json_schema_extra: dict[str, Any] | None = None

Configuration for a uni-pydantic field.

FieldConfig( index: Optional[Literal['btree', 'hash', 'fulltext', 'vector', 'sparse']] = None, unique: bool = False, tokenizer: str | None = None, metric: Optional[Literal['l2', 'cosine', 'dot']] = None, generated: str | None = None, default: Any = <factory>, default_factory: Callable[[], typing.Any] | None = None, alias: str | None = None, title: str | None = None, description: str | None = None, examples: list[typing.Any] | None = None, exclude: bool = False, json_schema_extra: dict[str, typing.Any] | None = None)

index: Optional[Literal['btree', 'hash', 'fulltext', 'vector', 'sparse']] = None

unique: bool = False

tokenizer: str | None = None

metric: Optional[Literal['l2', 'cosine', 'dot']] = None

generated: str | None = None

default: Any

default_factory: Callable[[], typing.Any] | None = None

alias: str | None = None

title: str | None = None

description: str | None = None

examples: list[typing.Any] | None = None

json_schema_extra: dict[str, typing.Any] | None = None

def Relationship( edge_type: str, *, direction: Literal['outgoing', 'incoming', 'both'] = 'outgoing', edge_model: type[UniEdge] | None = None, eager: bool = False, cascade_delete: bool = False) -> Any: View Source

272def Relationship(
273    edge_type: str,
274    *,
275    direction: Direction = "outgoing",
276    edge_model: type[UniEdge] | None = None,
277    eager: bool = False,
278    cascade_delete: bool = False,
279) -> Any:
280    """
281    Declare a relationship to another node type.
282
283    Relationships are lazy-loaded by default. Use eager=True or
284    query.eager_load() to load them with the parent query.
285
286    Args:
287        edge_type: The edge type name (e.g., "FRIEND_OF", "WORKS_AT").
288        direction: Relationship direction:
289            - "outgoing": Follow edges from this node (default)
290            - "incoming": Follow edges to this node
291            - "both": Follow edges in both directions
292        edge_model: Optional UniEdge subclass for typed edge properties.
293        eager: Whether to eager-load this relationship by default.
294        cascade_delete: Whether to delete related edges when this node is deleted.
295
296    Returns:
297        A RelationshipDescriptor that will be processed during model creation.
298
299    Examples:
300        >>> class Person(UniNode):
301        ...     # Outgoing relationship (default)
302        ...     follows: list["Person"] = Relationship("FOLLOWS")
303        ...
304        ...     # Incoming relationship
305        ...     followers: list["Person"] = Relationship("FOLLOWS", direction="incoming")
306        ...
307        ...     # Single optional relationship
308        ...     manager: "Person | None" = Relationship("REPORTS_TO")
309        ...
310        ...     # Relationship with edge properties
311        ...     friendships: list[tuple["Person", FriendshipEdge]] = Relationship(
312        ...         "FRIEND_OF",
313        ...         edge_model=FriendshipEdge
314        ...     )
315    """
316    config = RelationshipConfig(
317        edge_type=edge_type,
318        direction=direction,
319        edge_model=edge_model,
320        eager=eager,
321        cascade_delete=cascade_delete,
322    )
323    # Return a marker that will be processed by the metaclass
324    return _RelationshipMarker(config)

Declare a relationship to another node type.

Relationships are lazy-loaded by default. Use eager=True or query.eager_load() to load them with the parent query.

Args: edge_type: The edge type name (e.g., "FRIEND_OF", "WORKS_AT"). direction: Relationship direction: - "outgoing": Follow edges from this node (default) - "incoming": Follow edges to this node - "both": Follow edges in both directions edge_model: Optional UniEdge subclass for typed edge properties. eager: Whether to eager-load this relationship by default. cascade_delete: Whether to delete related edges when this node is deleted.

Returns: A RelationshipDescriptor that will be processed during model creation.

Examples:

class Person(UniNode): ... # Outgoing relationship (default) ... follows: list["Person"] = Relationship("FOLLOWS") ... ... # Incoming relationship ... followers: list["Person"] = Relationship("FOLLOWS", direction="incoming") ... ... # Single optional relationship ... manager: "Person | None" = Relationship("REPORTS_TO") ... ... # Relationship with edge properties ... friendships: list[tuple["Person", FriendshipEdge]] = Relationship( ... "FRIEND_OF", ... edge_model=FriendshipEdge ... )

@dataclass

class RelationshipConfig: View Source

185@dataclass
186class RelationshipConfig:
187    """Configuration for a relationship field."""
188
189    edge_type: str
190    direction: Direction = "outgoing"
191    edge_model: type[UniEdge] | None = None
192    eager: bool = False
193    cascade_delete: bool = False

Configuration for a relationship field.

RelationshipConfig( edge_type: str, direction: Literal['outgoing', 'incoming', 'both'] = 'outgoing', edge_model: type[UniEdge] | None = None, eager: bool = False, cascade_delete: bool = False)

edge_type: str

direction: Literal['outgoing', 'incoming', 'both'] = 'outgoing'

edge_model: type[UniEdge] | None = None

eager: bool = False

cascade_delete: bool = False

class RelationshipDescriptor(typing.Generic[~NodeT]): View Source

196class RelationshipDescriptor(Generic[NodeT]):
197    """
198    Descriptor for relationship fields that enables lazy loading.
199
200    When accessed on an instance, it returns the related nodes.
201    When accessed on the class, it returns the descriptor for query building.
202    """
203
204    def __init__(
205        self,
206        config: RelationshipConfig,
207        field_name: str,
208        target_type: type[NodeT] | str | None = None,
209        is_list: bool = True,
210    ) -> None:
211        self.config = config
212        self.field_name = field_name
213        self.target_type = target_type
214        self.is_list = is_list
215        self._cache_attr = f"_rel_cache_{field_name}"
216
217    def __set_name__(self, owner: type, name: str) -> None:
218        self.field_name = name
219        self._cache_attr = f"_rel_cache_{name}"
220
221    @overload
222    def __get__(
223        self, obj: None, objtype: type[NodeT]
224    ) -> RelationshipDescriptor[NodeT]: ...
225
226    @overload
227    def __get__(
228        self, obj: NodeT, objtype: type[NodeT] | None = None
229    ) -> list[NodeT] | NodeT | None: ...
230
231    def __get__(
232        self, obj: NodeT | None, objtype: type[NodeT] | None = None
233    ) -> RelationshipDescriptor[NodeT] | list[NodeT] | NodeT | None:
234        if obj is None:
235            # Class-level access returns the descriptor
236            return self
237
238        # Instance-level access - check cache first
239        if hasattr(obj, self._cache_attr):
240            cached = getattr(obj, self._cache_attr)
241            return cast("list[NodeT] | NodeT | None", cached)
242
243        # Check if we have a session for lazy loading
244        session = getattr(obj, "_session", None)
245        if session is None:
246            from .exceptions import LazyLoadError
247
248            raise LazyLoadError(
249                self.field_name,
250                "No session attached. Use session.get() or enable eager loading.",
251            )
252
253        # Lazy load the relationship
254        result = session._load_relationship(obj, self)
255
256        # Cache the result
257        setattr(obj, self._cache_attr, result)
258        return cast("list[NodeT] | NodeT | None", result)
259
260    def __set__(self, obj: NodeT, value: list[NodeT] | NodeT | None) -> None:
261        # Allow setting the cached value (e.g., during eager loading)
262        setattr(obj, self._cache_attr, value)
263
264    def __repr__(self) -> str:
265        return f"Relationship({self.config.edge_type!r}, direction={self.config.direction!r})"

Descriptor for relationship fields that enables lazy loading.

When accessed on an instance, it returns the related nodes. When accessed on the class, it returns the descriptor for query building.

RelationshipDescriptor( config: RelationshipConfig, field_name: str, target_type: type[~NodeT] | str | None = None, is_list: bool = True) View Source

204    def __init__(
205        self,
206        config: RelationshipConfig,
207        field_name: str,
208        target_type: type[NodeT] | str | None = None,
209        is_list: bool = True,
210    ) -> None:
211        self.config = config
212        self.field_name = field_name
213        self.target_type = target_type
214        self.is_list = is_list
215        self._cache_attr = f"_rel_cache_{field_name}"

config

field_name

target_type

is_list

def get_field_config( field_info: pydantic.fields.FieldInfo) -> FieldConfig | None: View Source

175def get_field_config(field_info: FieldInfo) -> FieldConfig | None:
176    """Extract uni-pydantic config from a Pydantic FieldInfo."""
177    extra = field_info.json_schema_extra
178    if isinstance(extra, dict):
179        config = extra.get("uni_config")
180        if isinstance(config, FieldConfig):
181            return config
182    return None

Extract uni-pydantic config from a Pydantic FieldInfo.

IndexType = typing.Literal['btree', 'hash', 'fulltext', 'vector', 'sparse']

Direction = typing.Literal['outgoing', 'incoming', 'both']

VectorMetric = typing.Literal['l2', 'cosine', 'dot']

class Btic: View Source

281class Btic:
282    """A BTIC temporal interval value for Uni graph database.
283
284    Construct from an ISO 8601-inspired string literal::
285
286        Btic("1985")
287        Btic("1985-03/2024-06")
288        Btic("~1985")           # approximate certainty
289        Btic("2020-03/")        # ongoing (unbounded hi)
290
291    Use as a Pydantic model field type::
292
293        class Event(UniNode):
294            when: Btic
295    """
296
297    def __init__(self, value: str | object) -> None:
298        if _PyBtic is None:
299            raise ImportError("uni_db is required for Btic type")
300        if isinstance(value, str):
301            self._inner = _PyBtic(value)
302        elif _PyBtic is not None and isinstance(value, _PyBtic):
303            self._inner = value
304        elif isinstance(value, Btic):
305            self._inner = value._inner
306        else:
307            raise TypeError(f"Expected str or Btic, got {type(value)}")
308
309    @property
310    def lo(self) -> int:
311        """Lower bound in milliseconds since epoch."""
312        return self._inner.lo
313
314    @property
315    def hi(self) -> int:
316        """Upper bound in milliseconds since epoch."""
317        return self._inner.hi
318
319    @property
320    def meta(self) -> int:
321        """Raw 64-bit metadata word."""
322        return self._inner.meta
323
324    @property
325    def lo_granularity(self) -> str:
326        """Lower bound granularity name."""
327        return self._inner.lo_granularity
328
329    @property
330    def hi_granularity(self) -> str:
331        """Upper bound granularity name."""
332        return self._inner.hi_granularity
333
334    @property
335    def lo_certainty(self) -> str:
336        """Lower bound certainty name."""
337        return self._inner.lo_certainty
338
339    @property
340    def hi_certainty(self) -> str:
341        """Upper bound certainty name."""
342        return self._inner.hi_certainty
343
344    @property
345    def duration_ms(self) -> int | None:
346        """Duration in milliseconds, or None if unbounded."""
347        return self._inner.duration_ms
348
349    @property
350    def is_instant(self) -> bool:
351        """True if the interval is exactly 1 millisecond wide."""
352        return self._inner.is_instant
353
354    @property
355    def is_unbounded(self) -> bool:
356        """True if either bound is infinite."""
357        return self._inner.is_unbounded
358
359    @property
360    def is_finite(self) -> bool:
361        """True if both bounds are finite."""
362        return self._inner.is_finite
363
364    def __repr__(self) -> str:
365        return f'Btic("{self._inner}")'
366
367    def __str__(self) -> str:
368        return str(self._inner)
369
370    def __eq__(self, other: object) -> bool:
371        if isinstance(other, Btic):
372            return self._inner == other._inner
373        return False
374
375    def __hash__(self) -> int:
376        return hash(self._inner)
377
378    @classmethod
379    def __get_pydantic_core_schema__(
380        cls, source_type: Any, handler: GetCoreSchemaHandler
381    ) -> CoreSchema:
382        """Make Btic compatible with Pydantic v2."""
383
384        def validate_btic(v: Any) -> Btic:
385            if isinstance(v, Btic):
386                return v
387            if isinstance(v, str):
388                return Btic(v)
389            if _PyBtic is not None and isinstance(v, _PyBtic):
390                return Btic(v)
391            raise TypeError(f"Expected str or Btic, got {type(v)}")
392
393        return core_schema.no_info_plain_validator_function(
394            validate_btic,
395            serialization=core_schema.plain_serializer_function_ser_schema(
396                lambda v: str(v._inner) if isinstance(v, Btic) else str(v),
397                info_arg=False,
398            ),
399        )

A BTIC temporal interval value for Uni graph database.

Construct from an ISO 8601-inspired string literal::

Btic("1985")
Btic("1985-03/2024-06")
Btic("~1985")           # approximate certainty
Btic("2020-03/")        # ongoing (unbounded hi)

Use as a Pydantic model field type::

class Event(UniNode):
    when: Btic

Btic(value: str | object) View Source

297    def __init__(self, value: str | object) -> None:
298        if _PyBtic is None:
299            raise ImportError("uni_db is required for Btic type")
300        if isinstance(value, str):
301            self._inner = _PyBtic(value)
302        elif _PyBtic is not None and isinstance(value, _PyBtic):
303            self._inner = value
304        elif isinstance(value, Btic):
305            self._inner = value._inner
306        else:
307            raise TypeError(f"Expected str or Btic, got {type(value)}")

lo: int View Source

309    @property
310    def lo(self) -> int:
311        """Lower bound in milliseconds since epoch."""
312        return self._inner.lo

Lower bound in milliseconds since epoch.

hi: int View Source

314    @property
315    def hi(self) -> int:
316        """Upper bound in milliseconds since epoch."""
317        return self._inner.hi

Upper bound in milliseconds since epoch.

meta: int View Source

319    @property
320    def meta(self) -> int:
321        """Raw 64-bit metadata word."""
322        return self._inner.meta

Raw 64-bit metadata word.

lo_granularity: str View Source

324    @property
325    def lo_granularity(self) -> str:
326        """Lower bound granularity name."""
327        return self._inner.lo_granularity

Lower bound granularity name.

hi_granularity: str View Source

329    @property
330    def hi_granularity(self) -> str:
331        """Upper bound granularity name."""
332        return self._inner.hi_granularity

Upper bound granularity name.

lo_certainty: str View Source

334    @property
335    def lo_certainty(self) -> str:
336        """Lower bound certainty name."""
337        return self._inner.lo_certainty

Lower bound certainty name.

hi_certainty: str View Source

339    @property
340    def hi_certainty(self) -> str:
341        """Upper bound certainty name."""
342        return self._inner.hi_certainty

Upper bound certainty name.

duration_ms: int | None View Source

344    @property
345    def duration_ms(self) -> int | None:
346        """Duration in milliseconds, or None if unbounded."""
347        return self._inner.duration_ms

Duration in milliseconds, or None if unbounded.

is_instant: bool View Source

349    @property
350    def is_instant(self) -> bool:
351        """True if the interval is exactly 1 millisecond wide."""
352        return self._inner.is_instant

True if the interval is exactly 1 millisecond wide.

is_unbounded: bool View Source

354    @property
355    def is_unbounded(self) -> bool:
356        """True if either bound is infinite."""
357        return self._inner.is_unbounded

True if either bound is infinite.

is_finite: bool View Source

359    @property
360    def is_finite(self) -> bool:
361        """True if both bounds are finite."""
362        return self._inner.is_finite

True if both bounds are finite.

class Vector(typing.Generic[~N]): View Source

 67class Vector(Generic[N], metaclass=VectorMeta):
 68    """
 69    A vector type with fixed dimensions for embeddings.
 70
 71    Usage:
 72        embedding: Vector[1536]  # 1536-dimensional vector
 73
 74    At runtime, vectors are stored as list[float].
 75    """
 76
 77    __dimensions__: int = 0
 78    __origin__: type | None = None
 79
 80    def __init__(self, values: list[float]) -> None:
 81        expected = self.__class__.__dimensions__
 82        if expected > 0 and len(values) != expected:
 83            raise ValueError(f"Vector expects {expected} dimensions, got {len(values)}")
 84        self._values = values
 85
 86    @property
 87    def values(self) -> list[float]:
 88        return self._values
 89
 90    def __repr__(self) -> str:
 91        dims = self.__class__.__dimensions__
 92        return (
 93            f"Vector[{dims}]({self._values[:3]}...)"
 94            if len(self._values) > 3
 95            else f"Vector[{dims}]({self._values})"
 96        )
 97
 98    def __eq__(self, other: object) -> bool:
 99        if isinstance(other, Vector):
100            return self._values == other._values
101        if isinstance(other, list):
102            return self._values == other
103        return False
104
105    def __len__(self) -> int:
106        return len(self._values)
107
108    def __iter__(self):  # type: ignore[no-untyped-def]
109        return iter(self._values)
110
111    @classmethod
112    def __get_pydantic_core_schema__(
113        cls, source_type: Any, handler: GetCoreSchemaHandler
114    ) -> CoreSchema:
115        """Make Vector compatible with Pydantic v2."""
116        dimensions = getattr(source_type, "__dimensions__", 0)
117        vec_cls = source_type if dimensions > 0 else cls
118
119        def validate_vector(v: Any) -> Vector:  # type: ignore[type-arg]
120            if isinstance(v, Vector):
121                if dimensions > 0 and len(v) != dimensions:
122                    raise ValueError(
123                        f"Vector expects {dimensions} dimensions, got {len(v)}"
124                    )
125                return v
126            if isinstance(v, list):
127                if dimensions > 0 and len(v) != dimensions:
128                    raise ValueError(
129                        f"Vector expects {dimensions} dimensions, got {len(v)}"
130                    )
131                return vec_cls([float(x) for x in v])
132            raise TypeError(f"Expected list or Vector, got {type(v)}")
133
134        return core_schema.no_info_plain_validator_function(
135            validate_vector,
136            serialization=core_schema.plain_serializer_function_ser_schema(
137                lambda v: v.values if isinstance(v, Vector) else list(v),
138                info_arg=False,
139            ),
140        )

A vector type with fixed dimensions for embeddings.

Usage: embedding: Vector[1536] # 1536-dimensional vector

At runtime, vectors are stored as list[float].

Vector(values: list[float]) View Source

80    def __init__(self, values: list[float]) -> None:
81        expected = self.__class__.__dimensions__
82        if expected > 0 and len(values) != expected:
83            raise ValueError(f"Vector expects {expected} dimensions, got {len(values)}")
84        self._values = values

values: list[float] View Source

86    @property
87    def values(self) -> list[float]:
88        return self._values

class SparseVector(typing.Generic[~N]): View Source

182class SparseVector(Generic[N], metaclass=SparseVectorMeta):
183    """
184    A learned-sparse (SPLADE / BGE-M3) vector over a fixed-size vocabulary.
185
186    Usage:
187        terms: SparseVector[30522]  # SPLADE head over a 30522-term BERT vocab
188
189    At runtime, holds parallel ``indices`` (term ids) and ``values`` (weights).
190    Accepts a ``dict[int, float]`` of term id -> weight, a ``uni_db.SparseVector``,
191    or an existing instance; ingestion serializes to the typed Rust binding when
192    available, otherwise to an ``{"indices": [...], "values": [...]}`` mapping.
193    """
194
195    __sparse_dimensions__: int = 0
196    __origin__: type | None = None
197
198    def __init__(self, indices: list[int], values: list[float]) -> None:
199        if len(indices) != len(values):
200            raise ValueError(
201                f"SparseVector indices/values length mismatch: "
202                f"{len(indices)} vs {len(values)}"
203            )
204        self._indices = [int(i) for i in indices]
205        self._values = [float(v) for v in values]
206
207    @property
208    def indices(self) -> list[int]:
209        return self._indices
210
211    @property
212    def values(self) -> list[float]:
213        return self._values
214
215    @classmethod
216    def from_dict(cls, mapping: dict[int, float]) -> SparseVector[Any]:
217        """Build from a ``{term_id: weight}`` mapping (sorted by term id)."""
218        items = sorted(mapping.items())
219        return cls([k for k, _ in items], [v for _, v in items])
220
221    def __repr__(self) -> str:
222        dims = self.__class__.__sparse_dimensions__
223        return f"SparseVector[{dims}](indices={self._indices}, values={self._values})"
224
225    def __eq__(self, other: object) -> bool:
226        if isinstance(other, SparseVector):
227            return self._indices == other._indices and self._values == other._values
228        return False
229
230    def __len__(self) -> int:
231        return len(self._indices)
232
233    @classmethod
234    def __get_pydantic_core_schema__(
235        cls, source_type: Any, handler: GetCoreSchemaHandler
236    ) -> CoreSchema:
237        """Make SparseVector compatible with Pydantic v2."""
238        dimensions = getattr(source_type, "__sparse_dimensions__", 0)
239        sv_cls = source_type if dimensions > 0 else cls
240
241        def validate_sparse(v: Any) -> SparseVector:  # type: ignore[type-arg]
242            if isinstance(v, SparseVector):
243                return v
244            if _PySparseVector is not None and isinstance(v, _PySparseVector):
245                return sv_cls(list(v.indices), list(v.values))
246            if isinstance(v, dict):
247                return sv_cls.from_dict(v)
248            if isinstance(v, (tuple, list)) and len(v) == 2:
249                return sv_cls(list(v[0]), list(v[1]))
250            raise TypeError(
251                f"Expected SparseVector, dict, or (indices, values), got {type(v)}"
252            )
253
254        def serialize_sparse(v: SparseVector) -> Any:  # type: ignore[type-arg]
255            if _PySparseVector is not None:
256                return _PySparseVector(v.indices, v.values)
257            return {"indices": v.indices, "values": v.values}
258
259        return core_schema.no_info_plain_validator_function(
260            validate_sparse,
261            serialization=core_schema.plain_serializer_function_ser_schema(
262                serialize_sparse,
263                info_arg=False,
264            ),
265        )

A learned-sparse (SPLADE / BGE-M3) vector over a fixed-size vocabulary.

Usage: terms: SparseVector[30522] # SPLADE head over a 30522-term BERT vocab

At runtime, holds parallel indices (term ids) and values (weights). Accepts a dict[int, float] of term id -> weight, a uni_db.SparseVector, or an existing instance; ingestion serializes to the typed Rust binding when available, otherwise to an {"indices": [...], "values": [...]} mapping.

SparseVector(indices: list[int], values: list[float]) View Source

198    def __init__(self, indices: list[int], values: list[float]) -> None:
199        if len(indices) != len(values):
200            raise ValueError(
201                f"SparseVector indices/values length mismatch: "
202                f"{len(indices)} vs {len(values)}"
203            )
204        self._indices = [int(i) for i in indices]
205        self._values = [float(v) for v in values]

indices: list[int] View Source

207    @property
208    def indices(self) -> list[int]:
209        return self._indices

values: list[float] View Source

211    @property
212    def values(self) -> list[float]:
213        return self._values

@classmethod

def from_dict(cls, mapping: dict[int, float]) -> 'SparseVector[Any]': View Source

215    @classmethod
216    def from_dict(cls, mapping: dict[int, float]) -> SparseVector[Any]:
217        """Build from a ``{term_id: weight}`` mapping (sorted by term id)."""
218        items = sorted(mapping.items())
219        return cls([k for k, _ in items], [v for _, v in items])

Build from a {term_id: weight} mapping (sorted by term id).

def python_type_to_uni(type_hint: Any, *, nullable: bool = False) -> tuple[str, bool]: View Source

558def python_type_to_uni(type_hint: Any, *, nullable: bool = False) -> tuple[str, bool]:
559    """
560    Convert a Python type hint to a Uni DataType string.
561
562    Args:
563        type_hint: The Python type hint to convert.
564        nullable: Whether the field is explicitly nullable.
565
566    Returns:
567        Tuple of (uni_data_type, is_nullable)
568
569    Raises:
570        TypeMappingError: If the type cannot be mapped.
571    """
572    # Unwrap Annotated if present
573    type_hint, _ = unwrap_annotated(type_hint)
574
575    # Check for optional (T | None)
576    is_opt, inner_type = is_optional(type_hint)
577    if is_opt:
578        uni_type, _ = python_type_to_uni(inner_type)
579        return uni_type, True
580
581    # Check for SparseVector types (before dense Vector: distinct marker attr).
582    sparse_dims = get_sparse_vector_dimensions(type_hint)
583    if sparse_dims is not None:
584        return f"sparse_vector:{sparse_dims}", nullable
585
586    # Check for Vector types
587    dims = get_vector_dimensions(type_hint)
588    if dims is not None:
589        return f"vector:{dims}", nullable
590
591    # Check for list types
592    is_lst, elem_type = is_list_type(type_hint)
593    if is_lst:
594        if elem_type in (str, int, float, bool):
595            # Simple list types
596            elem_uni = TYPE_MAP.get(elem_type, "string")
597            return f"list:{elem_uni}", nullable
598        # list[Vector[N]] (multi-vector / ColBERT) -> list:vector:N
599        elem_dims = get_vector_dimensions(elem_type)
600        if elem_dims is not None:
601            return f"list:vector:{elem_dims}", nullable
602        # Complex list types stored as JSON
603        return "json", nullable
604
605    # Direct type mapping
606    if type_hint in TYPE_MAP:
607        return TYPE_MAP[type_hint], nullable
608
609    # Handle generic dict types -> typed MAP<STRING, V> when the key is `str` and the
610    # value type maps to a concrete Uni type; otherwise schemaless JSON. Keys must be
611    # `str` because the storage value model (Value::Map) is string-keyed.
612    origin = get_origin(type_hint)
613    if origin is dict:
614        args = get_args(type_hint)
615        if len(args) == 2 and args[0] is str:
616            try:
617                val_uni, _ = python_type_to_uni(args[1])
618            except TypeMappingError:
619                val_uni = "json"
620            # Recurse handles nested values: dict[str, list[int]] -> map:string:list:int64,
621            # dict[str, Vector[N]] -> map:string:vector:N, dict[str, dict[str,int]] -> nested.
622            if val_uni != "json":
623                return f"map:string:{val_uni}", nullable
624        return "json", nullable
625
626    # Handle forward references (strings)
627    if isinstance(type_hint, str):
628        # This is a forward reference, can't resolve here
629        raise TypeMappingError(
630            type_hint,
631            f"Cannot resolve forward reference {type_hint!r}. "
632            "Ensure the referenced class is defined before schema sync.",
633        )
634
635    raise TypeMappingError(type_hint)

Convert a Python type hint to a Uni DataType string.

Args: type_hint: The Python type hint to convert. nullable: Whether the field is explicitly nullable.

Returns: Tuple of (uni_data_type, is_nullable)

Raises: TypeMappingError: If the type cannot be mapped.

def uni_to_python_type(uni_type: str) -> type: View Source

638def uni_to_python_type(uni_type: str) -> type:
639    """
640    Convert a Uni DataType string to a Python type.
641
642    Args:
643        uni_type: The Uni data type string.
644
645    Returns:
646        The corresponding Python type.
647    """
648    # Reverse mapping — manually constructed to avoid bytes overwriting str for "string"
649    _REVERSE_MAP: dict[str, type] = {
650        "string": str,
651        "int64": int,
652        "float64": float,
653        "bool": bool,
654        "datetime": datetime,
655        "date": date,
656        "time": time,
657        "duration": timedelta,
658        "json": dict,
659        "btic": Btic,
660        "bytes": bytes,
661    }
662
663    # Handle vector types
664    if uni_type.startswith("vector:"):
665        return list  # Vectors are stored as list[float]
666
667    # Handle list types
668    if uni_type.startswith("list:"):
669        return list
670
671    return _REVERSE_MAP.get(uni_type.lower(), str)

Convert a Uni DataType string to a Python type.

Args: uni_type: The Uni data type string.

Returns: The corresponding Python type.

def get_vector_dimensions(type_hint: Any) -> int | None: View Source

143def get_vector_dimensions(type_hint: Any) -> int | None:
144    """Extract vector dimensions from a Vector[N] type hint."""
145    if hasattr(type_hint, "__dimensions__"):
146        dims: int = type_hint.__dimensions__
147        return dims
148    origin = get_origin(type_hint)
149    if origin is Vector:
150        args = get_args(type_hint)
151        if args and isinstance(args[0], int):
152            return args[0]
153    return None

Extract vector dimensions from a Vector[N] type hint.

def get_sparse_vector_dimensions(type_hint: Any) -> int | None: View Source

268def get_sparse_vector_dimensions(type_hint: Any) -> int | None:
269    """Extract the vocabulary size from a SparseVector[N] type hint."""
270    if hasattr(type_hint, "__sparse_dimensions__"):
271        dims: int = type_hint.__sparse_dimensions__
272        return dims
273    origin = get_origin(type_hint)
274    if origin is SparseVector:
275        args = get_args(type_hint)
276        if args and isinstance(args[0], int):
277            return args[0]
278    return None

Extract the vocabulary size from a SparseVector[N] type hint.

def is_optional(type_hint: Any) -> tuple[bool, typing.Any]: View Source

402def is_optional(type_hint: Any) -> tuple[bool, Any]:
403    """
404    Check if a type hint is Optional (T | None).
405
406    Returns:
407        Tuple of (is_optional, inner_type)
408    """
409    origin = get_origin(type_hint)
410
411    # Handle Union types (including T | None which is Union[T, None])
412    if origin is Union:
413        args = get_args(type_hint)
414        non_none_args = [arg for arg in args if arg is not type(None)]
415        if len(non_none_args) == 1 and type(None) in args:
416            return True, non_none_args[0]
417
418    # Python 3.10+ uses types.UnionType for X | Y syntax
419    if isinstance(type_hint, types.UnionType):
420        args = get_args(type_hint)
421        non_none_args = [arg for arg in args if arg is not type(None)]
422        if len(non_none_args) == 1 and type(None) in args:
423            return True, non_none_args[0]
424
425    return False, type_hint

Check if a type hint is Optional (T | None).

Returns: Tuple of (is_optional, inner_type)

def is_list_type(type_hint: Any) -> tuple[bool, typing.Any | None]: View Source

428def is_list_type(type_hint: Any) -> tuple[bool, Any | None]:
429    """
430    Check if a type hint is a list type.
431
432    Returns:
433        Tuple of (is_list, element_type)
434    """
435    origin = get_origin(type_hint)
436    if origin is list:
437        args = get_args(type_hint)
438        return True, args[0] if args else Any
439    return False, None

Check if a type hint is a list type.

Returns: Tuple of (is_list, element_type)

def unwrap_annotated(type_hint: Any) -> tuple[typing.Any, tuple[typing.Any, ...]]: View Source

442def unwrap_annotated(type_hint: Any) -> tuple[Any, tuple[Any, ...]]:
443    """
444    Unwrap an Annotated type.
445
446    Returns:
447        Tuple of (base_type, metadata_tuple)
448    """
449    origin = get_origin(type_hint)
450    if origin is Annotated:
451        args = get_args(type_hint)
452        return args[0], args[1:]
453    return type_hint, ()

Unwrap an Annotated type.

Returns: Tuple of (base_type, metadata_tuple)

def python_to_db_value(value: Any, type_hint: Any) -> Any: View Source

460def python_to_db_value(value: Any, type_hint: Any) -> Any:
461    """Convert a Python value to a database-compatible value.
462
463    Passes datetime/date/time/timedelta through to the Rust layer which
464    converts them to proper Value::Temporal types. Converts Vector to
465    list[float] and passes through everything else.
466    """
467    if value is None:
468        return None
469
470    # list[Vector[N]] → list[list[float]] (multi-vector / ColBERT)
471    if isinstance(value, list) and value and isinstance(value[0], Vector):
472        return [v.values if isinstance(v, Vector) else v for v in value]
473
474    # Vector → list[float]
475    if isinstance(value, Vector):
476        return value.values
477
478    # Btic → unwrap to the Rust PyBtic for py_object_to_value
479    if isinstance(value, Btic):
480        return value._inner
481
482    # datetime/date/time/timedelta pass through — the Rust py_object_to_value
483    # handles conversion to Value::Temporal with proper type information.
484    return value

Convert a Python value to a database-compatible value.

Passes datetime/date/time/timedelta through to the Rust layer which converts them to proper Value::Temporal types. Converts Vector to list[float] and passes through everything else.

def db_to_python_value(value: Any, type_hint: Any) -> Any: View Source

487def db_to_python_value(value: Any, type_hint: Any) -> Any:
488    """Convert a database value back to a Python value.
489
490    The Rust layer now returns proper Python datetime/date/time objects
491    via Value::Temporal, so in most cases values pass through directly.
492    """
493    if value is None:
494        return None
495
496    # Unwrap Optional
497    _, inner = is_optional(type_hint)
498    if inner is not type_hint:
499        type_hint = inner
500
501    # Unwrap Annotated
502    type_hint, _ = unwrap_annotated(type_hint)
503
504    # If value is already the right Python type, pass through
505    if type_hint is datetime and isinstance(value, datetime):
506        return value
507    if type_hint is date and isinstance(value, date):
508        return value
509    if type_hint is time and isinstance(value, time):
510        return value
511    if type_hint is timedelta and isinstance(value, timedelta):
512        return value
513
514    # Btic — wrap Rust PyBtic in the pydantic Btic wrapper
515    if type_hint is Btic and _PyBtic is not None and isinstance(value, _PyBtic):
516        return Btic(value)
517
518    # Handle struct dict from Arrow deserialization (e.g. datetime struct)
519    if type_hint is datetime and isinstance(value, dict):
520        nanos = value.get("nanos_since_epoch")
521        if nanos is not None:
522            return datetime.fromtimestamp(nanos / 1_000_000_000)
523        return None
524
525    # list[Vector[N]] fields: list[list[float]] → list[Vector[N]] (multi-vector)
526    is_lst, elem_type = is_list_type(type_hint)
527    if is_lst:
528        elem_dims = get_vector_dimensions(elem_type)
529        if elem_dims is not None and isinstance(value, list):
530            vec_cls = Vector[elem_dims]
531            return [vec_cls(v) if isinstance(v, list) else v for v in value]
532
533    # Vector fields: list[float] → Vector
534    dims = get_vector_dimensions(type_hint)
535    if dims is not None and isinstance(value, list):
536        vec_cls = Vector[dims]
537        return vec_cls(value)
538
539    return value

Convert a database value back to a Python value.

The Rust layer now returns proper Python datetime/date/time objects via Value::Temporal, so in most cases values pass through directly.

DATETIME_TYPES = {<class 'datetime.date'>, <class 'datetime.datetime'>, <class 'datetime.timedelta'>, <class 'Btic'>, <class 'datetime.time'>}

class QueryBuilder(uni_pydantic.query._QueryBuilderBase[~NodeT]): View Source

1003class QueryBuilder(_QueryBuilderBase[NodeT]):
1004    """
1005    Immutable, type-safe query builder for graph queries.
1006
1007    Each method returns a **new** QueryBuilder instance. The original is
1008    never mutated. Provides a fluent API for building Cypher queries
1009    with type checking and IDE autocomplete support.
1010
1011    Example:
1012        >>> adults = (
1013        ...     session.query(Person)
1014        ...     .filter(Person.age >= 18)
1015        ...     .order_by(Person.name)
1016        ...     .limit(10)
1017        ...     .all()
1018        ... )
1019    """
1020
1021    def __init__(self, session: UniSession, model: type[NodeT]) -> None:
1022        self._init_state(session, model)
1023
1024    def _execute_query(
1025        self, cypher: str, params: dict[str, Any]
1026    ) -> list[dict[str, Any]]:
1027        """Execute a query, using query_with if timeout/max_memory is set."""
1028        if self._timeout is not None or self._max_memory is not None:
1029            builder = self._session._db_session.query_with(cypher)
1030            if params:
1031                builder = builder.params(params)
1032            if self._timeout is not None:
1033                builder = builder.timeout(self._timeout)
1034            if self._max_memory is not None:
1035                builder = builder.max_memory(self._max_memory)
1036            result = builder.fetch_all()
1037        else:
1038            result = self._session._db_session.query(cypher, params)
1039        return [row.to_dict() for row in result]
1040
1041    def all(self) -> list[NodeT]:
1042        """Execute the query and return all results."""
1043        cypher, params = self._build_cypher()
1044        results = self._execute_query(cypher, params)
1045        if self._is_search():
1046            instances = self._rows_to_scored_instances(results)
1047        else:
1048            instances = self._rows_to_instances(results)
1049        if self._eager_load and instances:
1050            self._session._eager_load_relationships(instances, self._eager_load)
1051        return instances
1052
1053    def first(self) -> NodeT | None:
1054        """Execute the query and return the first result."""
1055        clone = self._clone()
1056        clone._limit = 1
1057        results = clone.all()
1058        return results[0] if results else None
1059
1060    def one(self) -> NodeT:
1061        """Execute the query and return exactly one result.
1062
1063        Raises QueryError if no results or more than one result.
1064        """
1065        clone = self._clone()
1066        clone._limit = 2
1067        results = clone.all()
1068        if not results:
1069            raise QueryError("Query returned no results")
1070        if len(results) > 1:
1071            raise QueryError("Query returned more than one result")
1072        return results[0]
1073
1074    def count(self) -> int:
1075        """Execute the query and return the count of results."""
1076        cypher, params = self._build_count_cypher()
1077        results = self._execute_query(cypher, params)
1078        return cast(int, results[0]["count"]) if results else 0
1079
1080    def exists(self) -> bool:
1081        """Check if any matching records exist."""
1082        cypher, params = self._build_exists_cypher()
1083        results = self._execute_query(cypher, params)
1084        return len(results) > 0
1085
1086    def delete(self) -> int:
1087        """Delete all matching records (DETACH DELETE)."""
1088        cypher, params = self._build_delete_cypher()
1089        with self._session._db_session.tx() as tx:
1090            results = tx.query(cypher, params)
1091            tx.commit()
1092        return results[0].to_dict()["count"] if results else 0
1093
1094    def update(self, **kwargs: Any) -> int:
1095        """Update all matching records."""
1096        cypher, params = self._build_update_cypher(**kwargs)
1097        with self._session._db_session.tx() as tx:
1098            results = tx.query(cypher, params)
1099            tx.commit()
1100        return results[0].to_dict()["count"] if results else 0

Immutable, type-safe query builder for graph queries.

Each method returns a new QueryBuilder instance. The original is never mutated. Provides a fluent API for building Cypher queries with type checking and IDE autocomplete support.

Example:

adults = ( ... session.query(Person) ... .filter(Person.age >= 18) ... .order_by(Person.name) ... .limit(10) ... .all() ... )

QueryBuilder(session: UniSession, model: type[~NodeT]) View Source

1021    def __init__(self, session: UniSession, model: type[NodeT]) -> None:
1022        self._init_state(session, model)

def all(self) -> list[~NodeT]: View Source

1041    def all(self) -> list[NodeT]:
1042        """Execute the query and return all results."""
1043        cypher, params = self._build_cypher()
1044        results = self._execute_query(cypher, params)
1045        if self._is_search():
1046            instances = self._rows_to_scored_instances(results)
1047        else:
1048            instances = self._rows_to_instances(results)
1049        if self._eager_load and instances:
1050            self._session._eager_load_relationships(instances, self._eager_load)
1051        return instances

Execute the query and return all results.

def first(self) -> Optional[~NodeT]: View Source

1053    def first(self) -> NodeT | None:
1054        """Execute the query and return the first result."""
1055        clone = self._clone()
1056        clone._limit = 1
1057        results = clone.all()
1058        return results[0] if results else None

Execute the query and return the first result.

def one(self) -> ~NodeT: View Source

1060    def one(self) -> NodeT:
1061        """Execute the query and return exactly one result.
1062
1063        Raises QueryError if no results or more than one result.
1064        """
1065        clone = self._clone()
1066        clone._limit = 2
1067        results = clone.all()
1068        if not results:
1069            raise QueryError("Query returned no results")
1070        if len(results) > 1:
1071            raise QueryError("Query returned more than one result")
1072        return results[0]

Execute the query and return exactly one result.

Raises QueryError if no results or more than one result.

def count(self) -> int: View Source

1074    def count(self) -> int:
1075        """Execute the query and return the count of results."""
1076        cypher, params = self._build_count_cypher()
1077        results = self._execute_query(cypher, params)
1078        return cast(int, results[0]["count"]) if results else 0

Execute the query and return the count of results.

def exists(self) -> bool: View Source

1080    def exists(self) -> bool:
1081        """Check if any matching records exist."""
1082        cypher, params = self._build_exists_cypher()
1083        results = self._execute_query(cypher, params)
1084        return len(results) > 0

Check if any matching records exist.

def delete(self) -> int: View Source

1086    def delete(self) -> int:
1087        """Delete all matching records (DETACH DELETE)."""
1088        cypher, params = self._build_delete_cypher()
1089        with self._session._db_session.tx() as tx:
1090            results = tx.query(cypher, params)
1091            tx.commit()
1092        return results[0].to_dict()["count"] if results else 0

Delete all matching records (DETACH DELETE).

def update(self, **kwargs: Any) -> int: View Source

1094    def update(self, **kwargs: Any) -> int:
1095        """Update all matching records."""
1096        cypher, params = self._build_update_cypher(**kwargs)
1097        with self._session._db_session.tx() as tx:
1098            results = tx.query(cypher, params)
1099            tx.commit()
1100        return results[0].to_dict()["count"] if results else 0

Update all matching records.

class AsyncQueryBuilder(uni_pydantic.query._QueryBuilderBase[~NodeT]): View Source

 26class AsyncQueryBuilder(_QueryBuilderBase[NodeT]):
 27    """
 28    Immutable, async query builder for graph queries.
 29
 30    Inherits all Cypher-building and immutable builder methods from
 31    ``_QueryBuilderBase``. Only the execution methods are async.
 32    """
 33
 34    def __init__(self, session: AsyncUniSession, model: type[NodeT]) -> None:
 35        self._init_state(session, model)
 36
 37    async def _execute_query(
 38        self, cypher: str, params: dict[str, Any]
 39    ) -> list[dict[str, Any]]:
 40        """Execute a query, using query_with if timeout/max_memory is set."""
 41        if self._timeout is not None or self._max_memory is not None:
 42            builder = self._session._db_session.query_with(cypher)
 43            if params:
 44                builder = builder.params(params)
 45            if self._timeout is not None:
 46                builder = builder.timeout(self._timeout)
 47            if self._max_memory is not None:
 48                builder = builder.max_memory(self._max_memory)
 49            result = await builder.fetch_all()
 50        else:
 51            result = await self._session._db_session.query(cypher, params)
 52        return [row.to_dict() for row in result]
 53
 54    async def all(self) -> list[NodeT]:
 55        """Execute the query and return all results."""
 56        cypher, params = self._build_cypher()
 57        results = await self._execute_query(cypher, params)
 58        if self._is_search():
 59            instances = self._rows_to_scored_instances(results)
 60        else:
 61            instances = self._rows_to_instances(results)
 62        if self._eager_load and instances:
 63            await self._session._async_eager_load_relationships(
 64                instances, self._eager_load
 65            )
 66        return instances
 67
 68    async def first(self) -> NodeT | None:
 69        """Execute the query and return the first result."""
 70        clone = self._clone()
 71        clone._limit = 1
 72        results = await clone.all()
 73        return results[0] if results else None
 74
 75    async def one(self) -> NodeT:
 76        """Execute the query and return exactly one result.
 77
 78        Raises QueryError if no results or more than one result.
 79        """
 80        clone = self._clone()
 81        clone._limit = 2
 82        results = await clone.all()
 83        if not results:
 84            raise QueryError("Query returned no results")
 85        if len(results) > 1:
 86            raise QueryError("Query returned more than one result")
 87        return results[0]
 88
 89    async def count(self) -> int:
 90        """Execute the query and return the count of results."""
 91        cypher, params = self._build_count_cypher()
 92        results = await self._execute_query(cypher, params)
 93        return cast(int, results[0]["count"]) if results else 0
 94
 95    async def exists(self) -> bool:
 96        """Check if any matching records exist."""
 97        cypher, params = self._build_exists_cypher()
 98        results = await self._execute_query(cypher, params)
 99        return len(results) > 0
100
101    async def delete(self) -> int:
102        """Delete all matching records (DETACH DELETE)."""
103        cypher, params = self._build_delete_cypher()
104        async with await self._session._db_session.tx() as tx:
105            results = await tx.query(cypher, params)
106            await tx.commit()
107        return results[0].to_dict()["count"] if results else 0
108
109    async def update(self, **kwargs: Any) -> int:
110        """Update all matching records."""
111        cypher, params = self._build_update_cypher(**kwargs)
112        async with await self._session._db_session.tx() as tx:
113            results = await tx.query(cypher, params)
114            await tx.commit()
115        return results[0].to_dict()["count"] if results else 0

Immutable, async query builder for graph queries.

Inherits all Cypher-building and immutable builder methods from _QueryBuilderBase. Only the execution methods are async.

AsyncQueryBuilder( session: AsyncUniSession, model: type[~NodeT]) View Source

34    def __init__(self, session: AsyncUniSession, model: type[NodeT]) -> None:
35        self._init_state(session, model)

async def all(self) -> list[~NodeT]: View Source

54    async def all(self) -> list[NodeT]:
55        """Execute the query and return all results."""
56        cypher, params = self._build_cypher()
57        results = await self._execute_query(cypher, params)
58        if self._is_search():
59            instances = self._rows_to_scored_instances(results)
60        else:
61            instances = self._rows_to_instances(results)
62        if self._eager_load and instances:
63            await self._session._async_eager_load_relationships(
64                instances, self._eager_load
65            )
66        return instances

Execute the query and return all results.

async def first(self) -> Optional[~NodeT]: View Source

68    async def first(self) -> NodeT | None:
69        """Execute the query and return the first result."""
70        clone = self._clone()
71        clone._limit = 1
72        results = await clone.all()
73        return results[0] if results else None

Execute the query and return the first result.

async def one(self) -> ~NodeT: View Source

75    async def one(self) -> NodeT:
76        """Execute the query and return exactly one result.
77
78        Raises QueryError if no results or more than one result.
79        """
80        clone = self._clone()
81        clone._limit = 2
82        results = await clone.all()
83        if not results:
84            raise QueryError("Query returned no results")
85        if len(results) > 1:
86            raise QueryError("Query returned more than one result")
87        return results[0]

Execute the query and return exactly one result.

Raises QueryError if no results or more than one result.

async def count(self) -> int: View Source

89    async def count(self) -> int:
90        """Execute the query and return the count of results."""
91        cypher, params = self._build_count_cypher()
92        results = await self._execute_query(cypher, params)
93        return cast(int, results[0]["count"]) if results else 0

Execute the query and return the count of results.

async def exists(self) -> bool: View Source

95    async def exists(self) -> bool:
96        """Check if any matching records exist."""
97        cypher, params = self._build_exists_cypher()
98        results = await self._execute_query(cypher, params)
99        return len(results) > 0

Check if any matching records exist.

async def delete(self) -> int: View Source

101    async def delete(self) -> int:
102        """Delete all matching records (DETACH DELETE)."""
103        cypher, params = self._build_delete_cypher()
104        async with await self._session._db_session.tx() as tx:
105            results = await tx.query(cypher, params)
106            await tx.commit()
107        return results[0].to_dict()["count"] if results else 0

Delete all matching records (DETACH DELETE).

async def update(self, **kwargs: Any) -> int: View Source

109    async def update(self, **kwargs: Any) -> int:
110        """Update all matching records."""
111        cypher, params = self._build_update_cypher(**kwargs)
112        async with await self._session._db_session.tx() as tx:
113            results = await tx.query(cypher, params)
114            await tx.commit()
115        return results[0].to_dict()["count"] if results else 0

Update all matching records.

@dataclass

class FilterExpr: View Source

150@dataclass
151class FilterExpr:
152    """A filter expression for a query."""
153
154    property_name: str
155    op: FilterOp
156    value: Any = None
157
158    def to_cypher(self, node_var: str, param_name: str) -> tuple[str, dict[str, Any]]:
159        """Convert to Cypher WHERE clause fragment."""
160        prop = f"{node_var}.{self.property_name}"
161
162        if self.op == FilterOp.IS_NULL:
163            return f"{prop} IS NULL", {}
164        elif self.op == FilterOp.IS_NOT_NULL:
165            return f"{prop} IS NOT NULL", {}
166        elif self.op == FilterOp.IN:
167            return f"{prop} IN ${param_name}", {param_name: self.value}
168        elif self.op == FilterOp.NOT_IN:
169            return f"NOT {prop} IN ${param_name}", {param_name: self.value}
170        elif self.op == FilterOp.LIKE:
171            return f"{prop} =~ ${param_name}", {param_name: self.value}
172        elif self.op == FilterOp.STARTS_WITH:
173            return f"{prop} STARTS WITH ${param_name}", {param_name: self.value}
174        elif self.op == FilterOp.ENDS_WITH:
175            return f"{prop} ENDS WITH ${param_name}", {param_name: self.value}
176        elif self.op == FilterOp.CONTAINS:
177            return f"{prop} CONTAINS ${param_name}", {param_name: self.value}
178        else:
179            return f"{prop} {self.op.value} ${param_name}", {param_name: self.value}

A filter expression for a query.

FilterExpr( property_name: str, op: FilterOp, value: Any = None)

property_name: str

op: FilterOp

value: Any = None

def to_cypher( self, node_var: str, param_name: str) -> tuple[str, dict[str, typing.Any]]: View Source

158    def to_cypher(self, node_var: str, param_name: str) -> tuple[str, dict[str, Any]]:
159        """Convert to Cypher WHERE clause fragment."""
160        prop = f"{node_var}.{self.property_name}"
161
162        if self.op == FilterOp.IS_NULL:
163            return f"{prop} IS NULL", {}
164        elif self.op == FilterOp.IS_NOT_NULL:
165            return f"{prop} IS NOT NULL", {}
166        elif self.op == FilterOp.IN:
167            return f"{prop} IN ${param_name}", {param_name: self.value}
168        elif self.op == FilterOp.NOT_IN:
169            return f"NOT {prop} IN ${param_name}", {param_name: self.value}
170        elif self.op == FilterOp.LIKE:
171            return f"{prop} =~ ${param_name}", {param_name: self.value}
172        elif self.op == FilterOp.STARTS_WITH:
173            return f"{prop} STARTS WITH ${param_name}", {param_name: self.value}
174        elif self.op == FilterOp.ENDS_WITH:
175            return f"{prop} ENDS WITH ${param_name}", {param_name: self.value}
176        elif self.op == FilterOp.CONTAINS:
177            return f"{prop} CONTAINS ${param_name}", {param_name: self.value}
178        else:
179            return f"{prop} {self.op.value} ${param_name}", {param_name: self.value}

Convert to Cypher WHERE clause fragment.

class FilterOp(enum.Enum): View Source

131class FilterOp(Enum):
132    """Filter operation types."""
133
134    EQ = "="
135    NE = "<>"
136    LT = "<"
137    LE = "<="
138    GT = ">"
139    GE = ">="
140    IN = "IN"
141    NOT_IN = "NOT IN"
142    LIKE = "=~"
143    IS_NULL = "IS NULL"
144    IS_NOT_NULL = "IS NOT NULL"
145    STARTS_WITH = "STARTS WITH"
146    ENDS_WITH = "ENDS WITH"
147    CONTAINS = "CONTAINS"

Filter operation types.

EQ = <FilterOp.EQ: '='>

NE = <FilterOp.NE: '<>'>

LT = <FilterOp.LT: '<'>

LE = <FilterOp.LE: '<='>

GT = <FilterOp.GT: '>'>

GE = <FilterOp.GE: '>='>

IN = <FilterOp.IN: 'IN'>

NOT_IN = <FilterOp.NOT_IN: 'NOT IN'>

LIKE = <FilterOp.LIKE: '=~'>

IS_NULL = <FilterOp.IS_NULL: 'IS NULL'>

IS_NOT_NULL = <FilterOp.IS_NOT_NULL: 'IS NOT NULL'>

STARTS_WITH = <FilterOp.STARTS_WITH: 'STARTS WITH'>

ENDS_WITH = <FilterOp.ENDS_WITH: 'ENDS WITH'>

CONTAINS = <FilterOp.CONTAINS: 'CONTAINS'>

class ModelProxy(typing.Generic[~NodeT]): View Source

248class ModelProxy(Generic[NodeT]):
249    """
250    Proxy for model classes that provides property proxies.
251
252    Enables type-safe property access in query filters.
253
254    Example:
255        >>> Person.name  # Returns PropertyProxy for 'name'
256        >>> query.filter(Person.age >= 18)
257    """
258
259    def __init__(self, model: type[NodeT]) -> None:
260        self._model = model
261
262    def __getattr__(self, name: str) -> PropertyProxy[Any]:
263        if name.startswith("_"):
264            raise AttributeError(name)
265        return PropertyProxy(name, self._model)

Proxy for model classes that provides property proxies.

Enables type-safe property access in query filters.

Example:

Person.name # Returns PropertyProxy for 'name' query.filter(Person.age >= 18)

ModelProxy(model: type[~NodeT]) View Source

259    def __init__(self, model: type[NodeT]) -> None:
260        self._model = model

@dataclass

class OrderByClause: View Source

268@dataclass
269class OrderByClause:
270    """An ORDER BY clause."""
271
272    property_name: str
273    descending: bool = False

An ORDER BY clause.

OrderByClause(property_name: str, descending: bool = False)

property_name: str

descending: bool = False

@dataclass

class TraversalStep: View Source

276@dataclass
277class TraversalStep:
278    """A relationship traversal step."""
279
280    edge_type: str
281    direction: Literal["outgoing", "incoming", "both"]
282    target_label: str | None = None

A relationship traversal step.

TraversalStep( edge_type: str, direction: Literal['outgoing', 'incoming', 'both'], target_label: str | None = None)

edge_type: str

direction: Literal['outgoing', 'incoming', 'both']

target_label: str | None = None

@dataclass

class VectorSearchConfig: View Source

285@dataclass
286class VectorSearchConfig:
287    """Configuration for vector similarity search."""
288
289    property_name: str
290    query_vector: list[float]
291    k: int
292    threshold: float | None = None
293    pre_filter: str | None = None

Configuration for vector similarity search.

VectorSearchConfig( property_name: str, query_vector: list[float], k: int, threshold: float | None = None, pre_filter: str | None = None)

property_name: str

query_vector: list[float]

k: int

threshold: float | None = None

pre_filter: str | None = None

@dataclass

class SparseSearchConfig: View Source

296@dataclass
297class SparseSearchConfig:
298    """Configuration for learned-sparse (SPLADE) similarity search."""
299
300    property_name: str
301    query_indices: list[int]
302    query_values: list[float]
303    k: int
304    threshold: float | None = None
305    pre_filter: str | None = None

Configuration for learned-sparse (SPLADE) similarity search.

SparseSearchConfig( property_name: str, query_indices: list[int], query_values: list[float], k: int, threshold: float | None = None, pre_filter: str | None = None)

property_name: str

query_indices: list[int]

query_values: list[float]

k: int

threshold: float | None = None

pre_filter: str | None = None

@dataclass

class HybridSearchConfig: View Source

308@dataclass
309class HybridSearchConfig:
310    """Configuration for three-way fused hybrid search (``uni.search``).
311
312    Each ``*_property`` is ``None`` when that retrieval arm is off. ``query_text``
313    is the shared string that drives both FTS matching and dense auto-embed;
314    ``query_vector`` holds a precomputed dense vector (``None`` ⇒ auto-embed from
315    ``query_text``). ``sparse_query`` is the coerced ``(indices, values)`` pair
316    (``None`` ⇒ sparse arm off).
317    """
318
319    query_text: str
320    k: int
321    vector_property: str | None = None
322    fts_property: str | None = None
323    sparse_property: str | None = None
324    query_vector: list[float] | None = None
325    sparse_query: tuple[list[int], list[float]] | None = None
326    method: str = "rrf"
327    alpha: float | None = None
328    weights: list[float] | None = None
329    rrf_k: int | None = None
330    over_fetch: float | None = None
331    filter: str | None = None

Configuration for three-way fused hybrid search (uni.search).

Each *_property is None when that retrieval arm is off. query_text is the shared string that drives both FTS matching and dense auto-embed; query_vector holds a precomputed dense vector (None ⇒ auto-embed from query_text). sparse_query is the coerced (indices, values) pair (None ⇒ sparse arm off).

query_text: str

k: int

vector_property: str | None = None

fts_property: str | None = None

sparse_property: str | None = None

query_vector: list[float] | None = None

sparse_query: tuple[list[int], list[float]] | None = None

method: str = 'rrf'

alpha: float | None = None

weights: list[float] | None = None

rrf_k: int | None = None

over_fetch: float | None = None

filter: str | None = None

class SchemaGenerator: View Source

 66class SchemaGenerator:
 67    """Generates Uni database schema from registered models."""
 68
 69    def __init__(self) -> None:
 70        self._node_models: dict[str, type[UniNode]] = {}
 71        self._edge_models: dict[str, type[UniEdge]] = {}
 72        self._schema: DatabaseSchema | None = None
 73
 74    def register_node(self, model: type[UniNode]) -> None:
 75        """Register a node model for schema generation."""
 76        label = model.__label__
 77        if not label:
 78            raise SchemaError(f"Model {model.__name__} has no __label__", model)
 79        self._node_models[label] = model
 80        self._schema = None  # Invalidate cached schema
 81
 82    def register_edge(self, model: type[UniEdge]) -> None:
 83        """Register an edge model for schema generation."""
 84        edge_type = model.__edge_type__
 85        if not edge_type:
 86            raise SchemaError(f"Model {model.__name__} has no __edge_type__", model)
 87        self._edge_models[edge_type] = model
 88        self._schema = None
 89
 90    def register(self, *models: type[UniNode] | type[UniEdge]) -> None:
 91        """Register multiple models."""
 92        for model in models:
 93            if issubclass(model, UniEdge):
 94                self.register_edge(model)
 95            elif issubclass(model, UniNode):
 96                self.register_node(model)
 97            else:
 98                raise SchemaError(
 99                    f"Model {model.__name__} must be a subclass of UniNode or UniEdge"
100                )
101
102    def _generate_property_schema(
103        self,
104        model: type[UniNode] | type[UniEdge],
105        field_name: str,
106    ) -> PropertySchema:
107        """Generate schema for a single property field."""
108        field_info = model.model_fields[field_name]
109
110        # Get type hints with forward refs resolved
111        try:
112            hints = get_type_hints(model)
113            type_hint = hints.get(field_name, field_info.annotation)
114        except Exception:
115            type_hint = field_info.annotation
116
117        # Check for nullability
118        is_nullable, inner_type = is_optional(type_hint)
119
120        # Get Uni data type
121        data_type, nullable = python_type_to_uni(type_hint, nullable=is_nullable)
122
123        # Check for vector dimensions
124        vec_dims = get_vector_dimensions(inner_type if is_nullable else type_hint)
125        if vec_dims:
126            data_type = f"vector:{vec_dims}"
127
128        # Check for sparse-vector dimensions (vocabulary size)
129        sparse_dims = get_sparse_vector_dimensions(
130            inner_type if is_nullable else type_hint
131        )
132        if sparse_dims:
133            data_type = f"sparse_vector:{sparse_dims}"
134
135        # Get field config for index settings
136        config = get_field_config(field_info)
137        index_type = config.index if config else None
138        unique = config.unique if config else False
139        tokenizer = config.tokenizer if config else None
140        metric = config.metric if config else None
141
142        # Auto-create vector index for Vector fields (regardless of Field config)
143        if vec_dims and not index_type:
144            index_type = "vector"
145
146        # Auto-create sparse index for SparseVector fields
147        if sparse_dims and not index_type:
148            index_type = "sparse"
149
150        return PropertySchema(
151            name=field_name,
152            data_type=data_type,
153            nullable=nullable,
154            index_type=index_type,
155            unique=unique,
156            tokenizer=tokenizer,
157            metric=metric,
158        )
159
160    def _generate_label_schema(self, model: type[UniNode]) -> LabelSchema:
161        """Generate schema for a node model."""
162        label = model.__label__
163
164        properties = {}
165        for field_name in model.get_property_fields():
166            prop_schema = self._generate_property_schema(model, field_name)
167            properties[field_name] = prop_schema
168
169        return LabelSchema(
170            name=label,
171            properties=properties,
172        )
173
174    def _generate_edge_type_schema(self, model: type[UniEdge]) -> EdgeTypeSchema:
175        """Generate schema for an edge model."""
176        edge_type = model.__edge_type__
177        from_labels = model.get_from_labels()
178        to_labels = model.get_to_labels()
179
180        # If from/to not specified, allow any labels
181        if not from_labels:
182            from_labels = list(self._node_models.keys())
183        if not to_labels:
184            to_labels = list(self._node_models.keys())
185
186        properties = {}
187        for field_name in model.get_property_fields():
188            prop_schema = self._generate_property_schema(model, field_name)
189            properties[field_name] = prop_schema
190
191        return EdgeTypeSchema(
192            name=edge_type,
193            from_labels=from_labels,
194            to_labels=to_labels,
195            properties=properties,
196        )
197
198    def generate(self) -> DatabaseSchema:
199        """Generate the complete database schema."""
200        if self._schema is not None:
201            return self._schema
202
203        schema = DatabaseSchema()
204
205        # Generate label schemas
206        for label, model in self._node_models.items():
207            schema.labels[label] = self._generate_label_schema(model)
208
209        # Generate edge type schemas
210        for edge_type_name, edge_model in self._edge_models.items():
211            schema.edge_types[edge_type_name] = self._generate_edge_type_schema(
212                edge_model
213            )
214
215        # Edge types implied by `Relationship(...)` fields, for which no
216        # explicit `UniEdge` model was registered.
217        #
218        # These used to declare *every* registered label on both endpoints,
219        # discarding the two things the declaration actually states: the model
220        # that owns the field, and the model its annotation points at. That is
221        # not merely imprecise -- an edge type whose endpoints span several
222        # labels makes `properties(b)` on an unlabelled `b` come back empty, so
223        # every relationship target loaded as a propertyless node.
224        for model in self._node_models.values():
225            for rel_name, rel_config in model.get_relationship_fields().items():
226                edge_type = rel_config.edge_type
227                if edge_type in self._edge_models:
228                    # An explicit UniEdge model declares its own endpoints.
229                    continue
230
231                owner = model.__label__
232                target = self._relationship_target_label(model, rel_name)
233                if target is None:
234                    # Unresolvable annotation (e.g. a forward reference to a
235                    # model that was never registered). Fall back to the old
236                    # permissive shape rather than guessing wrong.
237                    src, dst = set(self._node_models), set(self._node_models)
238                elif rel_config.direction == "incoming":
239                    src, dst = {target}, {owner}
240                elif rel_config.direction == "both":
241                    src, dst = {owner, target}, {owner, target}
242                else:
243                    src, dst = {owner}, {target}
244
245                existing = schema.edge_types.get(edge_type)
246                if existing is None:
247                    schema.edge_types[edge_type] = EdgeTypeSchema(
248                        name=edge_type,
249                        from_labels=sorted(src),
250                        to_labels=sorted(dst),
251                    )
252                else:
253                    # Several models may declare the same edge type; the
254                    # endpoint sets union.
255                    existing.from_labels = sorted(set(existing.from_labels) | src)
256                    existing.to_labels = sorted(set(existing.to_labels) | dst)
257
258        self._schema = schema
259        return schema
260
261    def _relationship_target_label(self, model: type[Any], rel_name: str) -> str | None:
262        """Resolve the label a relationship field points at.
263
264        The descriptor carries `target_type` as either a model class or an
265        unresolved forward-reference string; a string is matched against
266        registered models by class name, since `__label__` may differ from it.
267        Returns None when the target cannot be resolved.
268        """
269        descriptor = getattr(model, rel_name, None)
270        target = getattr(descriptor, "target_type", None)
271        if target is None:
272            return None
273        if isinstance(target, str):
274            # A quoted annotation may be a bare name (`"Book"`) or a whole
275            # expression the metaclass could not evaluate (`"Bio | None"`,
276            # `"list[Book]"`). Pull identifiers out and match any of them
277            # against registered models, skipping typing spelling.
278            noise = {"None", "Optional", "Union", "list", "List", "set", "Set"}
279            for token in re.findall(r"[A-Za-z_][A-Za-z0-9_]*", target):
280                if token in noise:
281                    continue
282                for label, candidate in self._node_models.items():
283                    if candidate.__name__ == token or label == token:
284                        return label
285            return None
286        return getattr(target, "__label__", None)
287
288    def apply_to_database(self, db: uni_db.Uni) -> None:
289        """Apply the generated schema to a database using SchemaBuilder.
290
291        Uses db.schema() for atomic schema application with additive-only
292        semantics. Creates labels, edge types, properties, and indexes.
293        """
294        schema = self.generate()
295
296        # Build the full schema using SchemaBuilder, skipping existing labels/edge types
297        builder = db.schema()
298        has_changes = False
299
300        for label, label_schema in schema.labels.items():
301            if db.label_exists(label):
302                continue  # Additive-only: skip existing labels
303            lb = builder.label(label)
304            for prop in label_schema.properties.values():
305                # Check for vector type
306                if prop.data_type.startswith("vector:"):
307                    dims = int(prop.data_type.split(":")[1])
308                    lb = lb.vector(prop.name, dims)
309                elif prop.nullable:
310                    lb = lb.property_nullable(prop.name, prop.data_type)
311                else:
312                    lb = lb.property(prop.name, prop.data_type)
313
314                # Add indexes (not vector — vector is handled by .vector())
315                if prop.index_type and prop.index_type in ("btree", "hash"):
316                    lb = lb.index(prop.name, prop.index_type)
317            builder = lb.done()
318            has_changes = True
319
320        for edge_type, edge_schema in schema.edge_types.items():
321            if db.edge_type_exists(edge_type):
322                continue  # Skip existing edge types
323            eb = builder.edge_type(
324                edge_type, edge_schema.from_labels, edge_schema.to_labels
325            )
326            for prop in edge_schema.properties.values():
327                if prop.nullable:
328                    eb = eb.property_nullable(prop.name, prop.data_type)
329                else:
330                    eb = eb.property(prop.name, prop.data_type)
331            builder = eb.done()
332            has_changes = True
333
334        if has_changes:
335            builder.apply()
336
337        # Create vector and fulltext indexes via schema builder
338        for label, label_schema in schema.labels.items():
339            for prop in label_schema.properties.values():
340                if prop.index_type == "vector":
341                    metric = prop.metric or "l2"
342                    try:
343                        db.schema().label(label).index(
344                            prop.name, {"type": "vector", "metric": metric}
345                        ).apply()
346                    except Exception:
347                        pass  # Index may already exist
348                elif prop.index_type == "sparse":
349                    try:
350                        cfg = {"type": "sparse"}
351                        if prop.data_type.startswith("sparse_vector:"):
352                            cfg["dimensions"] = int(prop.data_type.split(":")[1])
353                        db.schema().label(label).index(prop.name, cfg).apply()
354                    except Exception:
355                        pass  # Index may already exist
356                elif prop.index_type == "fulltext":
357                    try:
358                        db.schema().label(label).index(prop.name, "fulltext").apply()
359                    except Exception:
360                        pass  # Index may already exist
361
362    async def async_apply_to_database(self, db: uni_db.AsyncUni) -> None:
363        """Apply the generated schema to an async database.
364
365        Async variant of apply_to_database using AsyncSchemaBuilder.
366        """
367        schema = self.generate()
368
369        # Build the full schema using AsyncSchemaBuilder, skipping existing labels/edge types
370        builder = db.schema()
371        has_changes = False
372
373        for label, label_schema in schema.labels.items():
374            if await db.label_exists(label):
375                continue
376            lb = builder.label(label)
377            for prop in label_schema.properties.values():
378                if prop.data_type.startswith("vector:"):
379                    dims = int(prop.data_type.split(":")[1])
380                    lb = lb.vector(prop.name, dims)
381                elif prop.nullable:
382                    lb = lb.property_nullable(prop.name, prop.data_type)
383                else:
384                    lb = lb.property(prop.name, prop.data_type)
385
386                if prop.index_type and prop.index_type in ("btree", "hash"):
387                    lb = lb.index(prop.name, prop.index_type)
388            builder = lb.done()
389            has_changes = True
390
391        for edge_type, edge_schema in schema.edge_types.items():
392            if await db.edge_type_exists(edge_type):
393                continue
394            eb = builder.edge_type(
395                edge_type, edge_schema.from_labels, edge_schema.to_labels
396            )
397            for prop in edge_schema.properties.values():
398                if prop.nullable:
399                    eb = eb.property_nullable(prop.name, prop.data_type)
400                else:
401                    eb = eb.property(prop.name, prop.data_type)
402            builder = eb.done()
403            has_changes = True
404
405        if has_changes:
406            await builder.apply()
407
408        # Create vector and fulltext indexes via schema builder
409        for label, label_schema in schema.labels.items():
410            for prop in label_schema.properties.values():
411                if prop.index_type == "vector":
412                    metric = prop.metric or "l2"
413                    try:
414                        await (
415                            db.schema()
416                            .label(label)
417                            .index(prop.name, {"type": "vector", "metric": metric})
418                            .apply()
419                        )
420                    except Exception:
421                        pass  # Index may already exist
422                elif prop.index_type == "sparse":
423                    try:
424                        cfg = {"type": "sparse"}
425                        if prop.data_type.startswith("sparse_vector:"):
426                            cfg["dimensions"] = int(prop.data_type.split(":")[1])
427                        await db.schema().label(label).index(prop.name, cfg).apply()
428                    except Exception:
429                        pass  # Index may already exist
430                elif prop.index_type == "fulltext":
431                    try:
432                        await (
433                            db.schema()
434                            .label(label)
435                            .index(prop.name, "fulltext")
436                            .apply()
437                        )
438                    except Exception:
439                        pass  # Index may already exist

Generates Uni database schema from registered models.

def register_node(self, model: type[UniNode]) -> None: View Source

74    def register_node(self, model: type[UniNode]) -> None:
75        """Register a node model for schema generation."""
76        label = model.__label__
77        if not label:
78            raise SchemaError(f"Model {model.__name__} has no __label__", model)
79        self._node_models[label] = model
80        self._schema = None  # Invalidate cached schema

def register_edge(self, model: type[UniEdge]) -> None: View Source

82    def register_edge(self, model: type[UniEdge]) -> None:
83        """Register an edge model for schema generation."""
84        edge_type = model.__edge_type__
85        if not edge_type:
86            raise SchemaError(f"Model {model.__name__} has no __edge_type__", model)
87        self._edge_models[edge_type] = model
88        self._schema = None

def register( self, *models: type[UniNode] | type[UniEdge]) -> None: View Source

 90    def register(self, *models: type[UniNode] | type[UniEdge]) -> None:
 91        """Register multiple models."""
 92        for model in models:
 93            if issubclass(model, UniEdge):
 94                self.register_edge(model)
 95            elif issubclass(model, UniNode):
 96                self.register_node(model)
 97            else:
 98                raise SchemaError(
 99                    f"Model {model.__name__} must be a subclass of UniNode or UniEdge"
100                )

def generate(self) -> DatabaseSchema: View Source

198    def generate(self) -> DatabaseSchema:
199        """Generate the complete database schema."""
200        if self._schema is not None:
201            return self._schema
202
203        schema = DatabaseSchema()
204
205        # Generate label schemas
206        for label, model in self._node_models.items():
207            schema.labels[label] = self._generate_label_schema(model)
208
209        # Generate edge type schemas
210        for edge_type_name, edge_model in self._edge_models.items():
211            schema.edge_types[edge_type_name] = self._generate_edge_type_schema(
212                edge_model
213            )
214
215        # Edge types implied by `Relationship(...)` fields, for which no
216        # explicit `UniEdge` model was registered.
217        #
218        # These used to declare *every* registered label on both endpoints,
219        # discarding the two things the declaration actually states: the model
220        # that owns the field, and the model its annotation points at. That is
221        # not merely imprecise -- an edge type whose endpoints span several
222        # labels makes `properties(b)` on an unlabelled `b` come back empty, so
223        # every relationship target loaded as a propertyless node.
224        for model in self._node_models.values():
225            for rel_name, rel_config in model.get_relationship_fields().items():
226                edge_type = rel_config.edge_type
227                if edge_type in self._edge_models:
228                    # An explicit UniEdge model declares its own endpoints.
229                    continue
230
231                owner = model.__label__
232                target = self._relationship_target_label(model, rel_name)
233                if target is None:
234                    # Unresolvable annotation (e.g. a forward reference to a
235                    # model that was never registered). Fall back to the old
236                    # permissive shape rather than guessing wrong.
237                    src, dst = set(self._node_models), set(self._node_models)
238                elif rel_config.direction == "incoming":
239                    src, dst = {target}, {owner}
240                elif rel_config.direction == "both":
241                    src, dst = {owner, target}, {owner, target}
242                else:
243                    src, dst = {owner}, {target}
244
245                existing = schema.edge_types.get(edge_type)
246                if existing is None:
247                    schema.edge_types[edge_type] = EdgeTypeSchema(
248                        name=edge_type,
249                        from_labels=sorted(src),
250                        to_labels=sorted(dst),
251                    )
252                else:
253                    # Several models may declare the same edge type; the
254                    # endpoint sets union.
255                    existing.from_labels = sorted(set(existing.from_labels) | src)
256                    existing.to_labels = sorted(set(existing.to_labels) | dst)
257
258        self._schema = schema
259        return schema

Generate the complete database schema.

def apply_to_database(self, db: Uni) -> None: View Source

288    def apply_to_database(self, db: uni_db.Uni) -> None:
289        """Apply the generated schema to a database using SchemaBuilder.
290
291        Uses db.schema() for atomic schema application with additive-only
292        semantics. Creates labels, edge types, properties, and indexes.
293        """
294        schema = self.generate()
295
296        # Build the full schema using SchemaBuilder, skipping existing labels/edge types
297        builder = db.schema()
298        has_changes = False
299
300        for label, label_schema in schema.labels.items():
301            if db.label_exists(label):
302                continue  # Additive-only: skip existing labels
303            lb = builder.label(label)
304            for prop in label_schema.properties.values():
305                # Check for vector type
306                if prop.data_type.startswith("vector:"):
307                    dims = int(prop.data_type.split(":")[1])
308                    lb = lb.vector(prop.name, dims)
309                elif prop.nullable:
310                    lb = lb.property_nullable(prop.name, prop.data_type)
311                else:
312                    lb = lb.property(prop.name, prop.data_type)
313
314                # Add indexes (not vector — vector is handled by .vector())
315                if prop.index_type and prop.index_type in ("btree", "hash"):
316                    lb = lb.index(prop.name, prop.index_type)
317            builder = lb.done()
318            has_changes = True
319
320        for edge_type, edge_schema in schema.edge_types.items():
321            if db.edge_type_exists(edge_type):
322                continue  # Skip existing edge types
323            eb = builder.edge_type(
324                edge_type, edge_schema.from_labels, edge_schema.to_labels
325            )
326            for prop in edge_schema.properties.values():
327                if prop.nullable:
328                    eb = eb.property_nullable(prop.name, prop.data_type)
329                else:
330                    eb = eb.property(prop.name, prop.data_type)
331            builder = eb.done()
332            has_changes = True
333
334        if has_changes:
335            builder.apply()
336
337        # Create vector and fulltext indexes via schema builder
338        for label, label_schema in schema.labels.items():
339            for prop in label_schema.properties.values():
340                if prop.index_type == "vector":
341                    metric = prop.metric or "l2"
342                    try:
343                        db.schema().label(label).index(
344                            prop.name, {"type": "vector", "metric": metric}
345                        ).apply()
346                    except Exception:
347                        pass  # Index may already exist
348                elif prop.index_type == "sparse":
349                    try:
350                        cfg = {"type": "sparse"}
351                        if prop.data_type.startswith("sparse_vector:"):
352                            cfg["dimensions"] = int(prop.data_type.split(":")[1])
353                        db.schema().label(label).index(prop.name, cfg).apply()
354                    except Exception:
355                        pass  # Index may already exist
356                elif prop.index_type == "fulltext":
357                    try:
358                        db.schema().label(label).index(prop.name, "fulltext").apply()
359                    except Exception:
360                        pass  # Index may already exist

Apply the generated schema to a database using SchemaBuilder.

Uses db.schema() for atomic schema application with additive-only semantics. Creates labels, edge types, properties, and indexes.

async def async_apply_to_database(self, db: AsyncUni) -> None: View Source

362    async def async_apply_to_database(self, db: uni_db.AsyncUni) -> None:
363        """Apply the generated schema to an async database.
364
365        Async variant of apply_to_database using AsyncSchemaBuilder.
366        """
367        schema = self.generate()
368
369        # Build the full schema using AsyncSchemaBuilder, skipping existing labels/edge types
370        builder = db.schema()
371        has_changes = False
372
373        for label, label_schema in schema.labels.items():
374            if await db.label_exists(label):
375                continue
376            lb = builder.label(label)
377            for prop in label_schema.properties.values():
378                if prop.data_type.startswith("vector:"):
379                    dims = int(prop.data_type.split(":")[1])
380                    lb = lb.vector(prop.name, dims)
381                elif prop.nullable:
382                    lb = lb.property_nullable(prop.name, prop.data_type)
383                else:
384                    lb = lb.property(prop.name, prop.data_type)
385
386                if prop.index_type and prop.index_type in ("btree", "hash"):
387                    lb = lb.index(prop.name, prop.index_type)
388            builder = lb.done()
389            has_changes = True
390
391        for edge_type, edge_schema in schema.edge_types.items():
392            if await db.edge_type_exists(edge_type):
393                continue
394            eb = builder.edge_type(
395                edge_type, edge_schema.from_labels, edge_schema.to_labels
396            )
397            for prop in edge_schema.properties.values():
398                if prop.nullable:
399                    eb = eb.property_nullable(prop.name, prop.data_type)
400                else:
401                    eb = eb.property(prop.name, prop.data_type)
402            builder = eb.done()
403            has_changes = True
404
405        if has_changes:
406            await builder.apply()
407
408        # Create vector and fulltext indexes via schema builder
409        for label, label_schema in schema.labels.items():
410            for prop in label_schema.properties.values():
411                if prop.index_type == "vector":
412                    metric = prop.metric or "l2"
413                    try:
414                        await (
415                            db.schema()
416                            .label(label)
417                            .index(prop.name, {"type": "vector", "metric": metric})
418                            .apply()
419                        )
420                    except Exception:
421                        pass  # Index may already exist
422                elif prop.index_type == "sparse":
423                    try:
424                        cfg = {"type": "sparse"}
425                        if prop.data_type.startswith("sparse_vector:"):
426                            cfg["dimensions"] = int(prop.data_type.split(":")[1])
427                        await db.schema().label(label).index(prop.name, cfg).apply()
428                    except Exception:
429                        pass  # Index may already exist
430                elif prop.index_type == "fulltext":
431                    try:
432                        await (
433                            db.schema()
434                            .label(label)
435                            .index(prop.name, "fulltext")
436                            .apply()
437                        )
438                    except Exception:
439                        pass  # Index may already exist

Apply the generated schema to an async database.

Async variant of apply_to_database using AsyncSchemaBuilder.

@dataclass

class DatabaseSchema: View Source

58@dataclass
59class DatabaseSchema:
60    """Complete database schema generated from models."""
61
62    labels: dict[str, LabelSchema] = field(default_factory=dict)
63    edge_types: dict[str, EdgeTypeSchema] = field(default_factory=dict)

Complete database schema generated from models.

DatabaseSchema( labels: dict[str, LabelSchema] = <factory>, edge_types: dict[str, EdgeTypeSchema] = <factory>)

labels: dict[str, LabelSchema]

edge_types: dict[str, EdgeTypeSchema]

@dataclass

class LabelSchema: View Source

40@dataclass
41class LabelSchema:
42    """Schema for a vertex label."""
43
44    name: str
45    properties: dict[str, PropertySchema] = field(default_factory=dict)

Schema for a vertex label.

LabelSchema( name: str, properties: dict[str, PropertySchema] = <factory>)

properties: dict[str, PropertySchema]

@dataclass

class EdgeTypeSchema: View Source

48@dataclass
49class EdgeTypeSchema:
50    """Schema for an edge type."""
51
52    name: str
53    from_labels: list[str] = field(default_factory=list)
54    to_labels: list[str] = field(default_factory=list)
55    properties: dict[str, PropertySchema] = field(default_factory=dict)

Schema for an edge type.

EdgeTypeSchema( name: str, from_labels: list[str] = <factory>, to_labels: list[str] = <factory>, properties: dict[str, PropertySchema] = <factory>)

from_labels: list[str]

to_labels: list[str]

properties: dict[str, PropertySchema]

@dataclass

class PropertySchema: View Source

27@dataclass
28class PropertySchema:
29    """Schema for a single property."""
30
31    name: str
32    data_type: str
33    nullable: bool = False
34    index_type: str | None = None
35    unique: bool = False
36    tokenizer: str | None = None
37    metric: str | None = None

Schema for a single property.

PropertySchema( name: str, data_type: str, nullable: bool = False, index_type: str | None = None, unique: bool = False, tokenizer: str | None = None, metric: str | None = None)

data_type: str

nullable: bool = False

index_type: str | None = None

unique: bool = False

tokenizer: str | None = None

metric: str | None = None

def generate_schema( *models: type[UniNode] | type[UniEdge]) -> DatabaseSchema: View Source

442def generate_schema(*models: type[UniNode] | type[UniEdge]) -> DatabaseSchema:
443    """Generate a database schema from the given models."""
444    generator = SchemaGenerator()
445    generator.register(*models)
446    return generator.generate()

Generate a database schema from the given models.

class AsyncUniDatabase: View Source

 78class AsyncUniDatabase:
 79    """
 80    Thin wrapper around uni-db AsyncUniBuilder for ergonomic async database creation.
 81
 82    Example:
 83        >>> db = await AsyncUniDatabase.open("./path").build()
 84        >>> db = await AsyncUniDatabase.temporary().build()
 85    """
 86
 87    def __init__(self, builder: uni_db.AsyncUniBuilder) -> None:
 88        self._builder = builder
 89
 90    @classmethod
 91    def open(cls, path: str) -> AsyncUniDatabase:
 92        """Open or create a database at the given path."""
 93        import uni_db
 94
 95        return cls(uni_db.AsyncUniBuilder.open(path))
 96
 97    @classmethod
 98    def temporary(cls) -> AsyncUniDatabase:
 99        """Create an ephemeral in-memory database."""
100        import uni_db
101
102        return cls(uni_db.AsyncUniBuilder.temporary())
103
104    @classmethod
105    def in_memory(cls) -> AsyncUniDatabase:
106        """Create a persistent in-memory database."""
107        import uni_db
108
109        return cls(uni_db.AsyncUniBuilder.in_memory())
110
111    def cache_size(self, bytes_: int) -> AsyncUniDatabase:
112        """Set the cache size in bytes."""
113        self._builder = self._builder.cache_size(bytes_)
114        return self
115
116    def parallelism(self, n: int) -> AsyncUniDatabase:
117        """Set the parallelism level."""
118        self._builder = self._builder.parallelism(n)
119        return self
120
121    async def build(self) -> uni_db.AsyncUni:
122        """Build and return the async database instance."""
123        return await self._builder.build()

Thin wrapper around uni-db AsyncUniBuilder for ergonomic async database creation.

Example:

db = await AsyncUniDatabase.open("./path").build() db = await AsyncUniDatabase.temporary().build()

AsyncUniDatabase(builder: AsyncUniBuilder) View Source

87    def __init__(self, builder: uni_db.AsyncUniBuilder) -> None:
88        self._builder = builder

@classmethod

def open(cls, path: str) -> AsyncUniDatabase: View Source

90    @classmethod
91    def open(cls, path: str) -> AsyncUniDatabase:
92        """Open or create a database at the given path."""
93        import uni_db
94
95        return cls(uni_db.AsyncUniBuilder.open(path))

Open or create a database at the given path.

@classmethod

def temporary(cls) -> AsyncUniDatabase: View Source

 97    @classmethod
 98    def temporary(cls) -> AsyncUniDatabase:
 99        """Create an ephemeral in-memory database."""
100        import uni_db
101
102        return cls(uni_db.AsyncUniBuilder.temporary())

Create an ephemeral in-memory database.

@classmethod

def in_memory(cls) -> AsyncUniDatabase: View Source

104    @classmethod
105    def in_memory(cls) -> AsyncUniDatabase:
106        """Create a persistent in-memory database."""
107        import uni_db
108
109        return cls(uni_db.AsyncUniBuilder.in_memory())

Create a persistent in-memory database.

def cache_size(self, bytes_: int) -> AsyncUniDatabase: View Source

111    def cache_size(self, bytes_: int) -> AsyncUniDatabase:
112        """Set the cache size in bytes."""
113        self._builder = self._builder.cache_size(bytes_)
114        return self

Set the cache size in bytes.

def parallelism(self, n: int) -> AsyncUniDatabase: View Source

116    def parallelism(self, n: int) -> AsyncUniDatabase:
117        """Set the parallelism level."""
118        self._builder = self._builder.parallelism(n)
119        return self

Set the parallelism level.

async def build(self) -> AsyncUni: View Source

121    async def build(self) -> uni_db.AsyncUni:
122        """Build and return the async database instance."""
123        return await self._builder.build()

Build and return the async database instance.

def before_create(func: ~F) -> ~F: View Source

40def before_create(func: F) -> F:
41    """
42    Mark a method to be called before the entity is created in the database.
43
44    The method is called after validation but before the INSERT operation.
45    Useful for setting timestamps, generating IDs, or final validation.
46
47    Example:
48        >>> class Person(UniNode):
49        ...     name: str
50        ...     created_at: datetime | None = None
51        ...
52        ...     @before_create
53        ...     def set_created_at(self):
54        ...         self.created_at = datetime.now()
55    """
56    return _mark_hook(_BEFORE_CREATE)(func)

Mark a method to be called before the entity is created in the database.

The method is called after validation but before the INSERT operation. Useful for setting timestamps, generating IDs, or final validation.

Example:

class Person(UniNode): ... name: str ... created_at: datetime | None = None ... ... @before_create ... def set_created_at(self): ... self.created_at = datetime.now()

def after_create(func: ~F) -> ~F: View Source

59def after_create(func: F) -> F:
60    """
61    Mark a method to be called after the entity is created in the database.
62
63    The method is called after the INSERT operation completes successfully.
64    The entity will have its vid/eid assigned at this point.
65
66    Example:
67        >>> class Person(UniNode):
68        ...     name: str
69        ...
70        ...     @after_create
71        ...     def log_creation(self):
72        ...         logger.info(f"Created person {self.name} with vid={self.vid}")
73    """
74    return _mark_hook(_AFTER_CREATE)(func)

Mark a method to be called after the entity is created in the database.

The method is called after the INSERT operation completes successfully. The entity will have its vid/eid assigned at this point.

Example:

class Person(UniNode): ... name: str ... ... @after_create ... def log_creation(self): ... logger.info(f"Created person {self.name} with vid={self.vid}")

def before_update(func: ~F) -> ~F: View Source

77def before_update(func: F) -> F:
78    """
79    Mark a method to be called before the entity is updated in the database.
80
81    The method is called before the UPDATE operation.
82    Useful for validation or updating timestamps.
83
84    Example:
85        >>> class Person(UniNode):
86        ...     name: str
87        ...     updated_at: datetime | None = None
88        ...
89        ...     @before_update
90        ...     def validate_and_timestamp(self):
91        ...         if not self.name:
92        ...             raise ValueError("Name cannot be empty")
93        ...         self.updated_at = datetime.now()
94    """
95    return _mark_hook(_BEFORE_UPDATE)(func)

Mark a method to be called before the entity is updated in the database.

The method is called before the UPDATE operation. Useful for validation or updating timestamps.

Example:

class Person(UniNode): ... name: str ... updated_at: datetime | None = None ... ... @before_update ... def validate_and_timestamp(self): ... if not self.name: ... raise ValueError("Name cannot be empty") ... self.updated_at = datetime.now()

def after_update(func: ~F) -> ~F: View Source

 98def after_update(func: F) -> F:
 99    """
100    Mark a method to be called after the entity is updated in the database.
101
102    The method is called after the UPDATE operation completes successfully.
103
104    Example:
105        >>> class Person(UniNode):
106        ...     name: str
107        ...
108        ...     @after_update
109        ...     def notify_change(self):
110        ...         events.emit("person_updated", self.vid)
111    """
112    return _mark_hook(_AFTER_UPDATE)(func)

Mark a method to be called after the entity is updated in the database.

The method is called after the UPDATE operation completes successfully.

Example:

class Person(UniNode): ... name: str ... ... @after_update ... def notify_change(self): ... events.emit("person_updated", self.vid)

def before_delete(func: ~F) -> ~F: View Source

115def before_delete(func: F) -> F:
116    """
117    Mark a method to be called before the entity is deleted from the database.
118
119    The method is called before the DELETE operation.
120    Useful for cleanup or validation.
121
122    Example:
123        >>> class Person(UniNode):
124        ...     name: str
125        ...
126        ...     @before_delete
127        ...     def cleanup(self):
128        ...         # Remove related data
129        ...         pass
130    """
131    return _mark_hook(_BEFORE_DELETE)(func)

Mark a method to be called before the entity is deleted from the database.

The method is called before the DELETE operation. Useful for cleanup or validation.

Example:

class Person(UniNode): ... name: str ... ... @before_delete ... def cleanup(self): ... # Remove related data ... pass

def after_delete(func: ~F) -> ~F: View Source

134def after_delete(func: F) -> F:
135    """
136    Mark a method to be called after the entity is deleted from the database.
137
138    The method is called after the DELETE operation completes successfully.
139    The entity's vid/eid will be cleared at this point.
140
141    Example:
142        >>> class Person(UniNode):
143        ...     name: str
144        ...
145        ...     @after_delete
146        ...     def log_deletion(self):
147        ...         logger.info(f"Deleted person {self.name}")
148    """
149    return _mark_hook(_AFTER_DELETE)(func)

Mark a method to be called after the entity is deleted from the database.

The method is called after the DELETE operation completes successfully. The entity's vid/eid will be cleared at this point.

Example:

class Person(UniNode): ... name: str ... ... @after_delete ... def log_deletion(self): ... logger.info(f"Deleted person {self.name}")

def before_load(func: ~F) -> ~F: View Source

152def before_load(func: F) -> F:
153    """
154    Mark a method to be called before the entity is loaded from the database.
155
156    This is a class method that receives the raw property dictionary.
157    Can be used to transform data before model instantiation.
158
159    Example:
160        >>> class Person(UniNode):
161        ...     name: str
162        ...
163        ...     @classmethod
164        ...     @before_load
165        ...     def transform_data(cls, props: dict) -> dict:
166        ...         # Normalize name
167        ...         if 'name' in props:
168        ...             props['name'] = props['name'].strip()
169        ...         return props
170    """
171    return _mark_hook(_BEFORE_LOAD)(func)

Mark a method to be called before the entity is loaded from the database.

This is a class method that receives the raw property dictionary. Can be used to transform data before model instantiation.

Example:

class Person(UniNode): ... name: str ... ... @classmethod ... @before_load ... def transform_data(cls, props: dict) -> dict: ... # Normalize name ... if 'name' in props: ... props['name'] = props['name'].strip() ... return props

def after_load(func: ~F) -> ~F: View Source

174def after_load(func: F) -> F:
175    """
176    Mark a method to be called after the entity is loaded from the database.
177
178    The method is called after the entity is instantiated from database data.
179    Useful for computing derived values or initializing non-persisted state.
180
181    Example:
182        >>> class Person(UniNode):
183        ...     first_name: str
184        ...     last_name: str
185        ...     full_name: str | None = None
186        ...
187        ...     @after_load
188        ...     def compute_full_name(self):
189        ...         self.full_name = f"{self.first_name} {self.last_name}"
190    """
191    return _mark_hook(_AFTER_LOAD)(func)

Mark a method to be called after the entity is loaded from the database.

The method is called after the entity is instantiated from database data. Useful for computing derived values or initializing non-persisted state.

Example:

class Person(UniNode): ... first_name: str ... last_name: str ... full_name: str | None = None ... ... @after_load ... def compute_full_name(self): ... self.full_name = f"{self.first_name} {self.last_name}"

class UniPydanticError(builtins.Exception): View Source

15class UniPydanticError(Exception):
16    """Base exception for all uni-pydantic errors."""

Base exception for all uni-pydantic errors.

class SchemaError(uni_pydantic.UniPydanticError): View Source

19class SchemaError(UniPydanticError):
20    """Error related to schema definition or generation."""
21
22    def __init__(self, message: str, model: type | None = None) -> None:
23        self.model = model
24        super().__init__(message)

Error related to schema definition or generation.

SchemaError(message: str, model: type | None = None) View Source

22    def __init__(self, message: str, model: type | None = None) -> None:
23        self.model = model
24        super().__init__(message)

model

class TypeMappingError(uni_pydantic.SchemaError): View Source

27class TypeMappingError(SchemaError):
28    """Error mapping Python type to Uni DataType."""
29
30    def __init__(self, python_type: Any, message: str | None = None) -> None:
31        self.python_type = python_type
32        msg = message or f"Cannot map Python type {python_type!r} to Uni DataType"
33        super().__init__(msg)

Error mapping Python type to Uni DataType.

TypeMappingError(python_type: Any, message: str | None = None) View Source

30    def __init__(self, python_type: Any, message: str | None = None) -> None:
31        self.python_type = python_type
32        msg = message or f"Cannot map Python type {python_type!r} to Uni DataType"
33        super().__init__(msg)

python_type

class ValidationError(uni_pydantic.UniPydanticError): View Source

36class ValidationError(UniPydanticError):
37    """Validation error for model instances."""

Validation error for model instances.

class SessionError(uni_pydantic.UniPydanticError): View Source

40class SessionError(UniPydanticError):
41    """Error related to session operations."""

Error related to session operations.

class NotRegisteredError(uni_pydantic.SessionError): View Source

44class NotRegisteredError(SessionError):
45    """Model type not registered with session."""
46
47    def __init__(self, model: type[UniNode] | type[UniEdge]) -> None:
48        self.model = model
49        super().__init__(
50            f"Model {model.__name__!r} is not registered with this session. "
51            f"Call session.register({model.__name__}) first."
52        )

Model type not registered with session.

NotRegisteredError( model: type[UniNode] | type[UniEdge]) View Source

47    def __init__(self, model: type[UniNode] | type[UniEdge]) -> None:
48        self.model = model
49        super().__init__(
50            f"Model {model.__name__!r} is not registered with this session. "
51            f"Call session.register({model.__name__}) first."
52        )

model

class NotPersisted(uni_pydantic.SessionError): View Source

55class NotPersisted(SessionError):
56    """Operation requires a persisted entity."""
57
58    def __init__(self, entity: UniNode | UniEdge) -> None:
59        self.entity = entity
60        super().__init__(
61            f"Entity {entity!r} is not persisted. Call session.add() and commit() first."
62        )

Operation requires a persisted entity.

NotPersisted(entity: UniNode | UniEdge) View Source

58    def __init__(self, entity: UniNode | UniEdge) -> None:
59        self.entity = entity
60        super().__init__(
61            f"Entity {entity!r} is not persisted. Call session.add() and commit() first."
62        )

entity

class NotTrackedError(uni_pydantic.SessionError): View Source

65class NotTrackedError(SessionError):
66    """Entity is not tracked by this session."""

Entity is not tracked by this session.

class TransactionError(uni_pydantic.SessionError): View Source

69class TransactionError(SessionError):
70    """Error related to transaction operations."""

Error related to transaction operations.

class QueryError(uni_pydantic.UniPydanticError): View Source

73class QueryError(UniPydanticError):
74    """Error executing a query."""

Error executing a query.

class RelationshipError(uni_pydantic.UniPydanticError): View Source

77class RelationshipError(UniPydanticError):
78    """Error related to relationship operations."""

Error related to relationship operations.

class LazyLoadError(uni_pydantic.RelationshipError): View Source

81class LazyLoadError(RelationshipError):
82    """Error lazy-loading a relationship."""
83
84    def __init__(self, field_name: str, reason: str) -> None:
85        self.field_name = field_name
86        super().__init__(f"Cannot lazy-load relationship '{field_name}': {reason}")

Error lazy-loading a relationship.

LazyLoadError(field_name: str, reason: str) View Source

84    def __init__(self, field_name: str, reason: str) -> None:
85        self.field_name = field_name
86        super().__init__(f"Cannot lazy-load relationship '{field_name}': {reason}")

field_name

class BulkLoadError(uni_pydantic.UniPydanticError): View Source

89class BulkLoadError(UniPydanticError):
90    """Error during bulk loading operations."""

Error during bulk loading operations.

class CypherInjectionError(uni_pydantic.QueryError): View Source

93class CypherInjectionError(QueryError):
94    """Property name validation failure — potential Cypher injection."""
95
96    def __init__(self, name: str, reason: str | None = None) -> None:
97        self.name = name
98        msg = reason or f"Invalid property name {name!r}: possible Cypher injection"
99        super().__init__(msg)

Property name validation failure — potential Cypher injection.

CypherInjectionError(name: str, reason: str | None = None) View Source

96    def __init__(self, name: str, reason: str | None = None) -> None:
97        self.name = name
98        msg = reason or f"Invalid property name {name!r}: possible Cypher injection"
99        super().__init__(msg)

name