surorm 0.0.20

No project description provided

surorm

A typed Python ORM and query builder for SurrealDB, built on Pydantic v2.

Status: pre-release (0.0.x). The API is not stable yet and breaking changes should be expected between versions — there is no compatibility shim between releases.

Features

• SurrealQL-native data types — String, Int, Float, Decimal, Boolean, Datetime, Duration, Bytes, Array[T], Set[T], Object, Json, Option[T], Range, RecordID, RecordLink[T], UUID, File. Each type both declares a field's schema and knows how to serialize/deserialize itself to/from SurrealQL.
• Pydantic-backed models — Model subclasses are real Pydantic v2 models (validation, mutation, model_dump(), etc. all work as expected) with automatic dirty tracking.
• Immutable, chainable query builder — Select, Create, Update, Delete, Relate, Define*, Remove, Transaction mirror SurrealQL directly. Every builder method returns a new instance, so partially-built queries can be safely reused and extended.
• Repository pattern — a small Repository[Model] generic gives you get, list_, save, update, and delete without hand-writing statements for common CRUD.
• Graph relations — declare edge tables with Relation and build RELATE ... -> ... -> ... statements directly.

Installation

pip install surorm

Requires Python 3.12+ and a running SurrealDB instance.

Quickstart

Define a model

from surorm import Model
from surorm.data_model import String, Int, Option

class User(Model):
    __table__ = 'user'

    name: String
    age: Int
    bio: Option[String] = None

Model extends Pydantic's BaseModel, so instances are constructed and validated the normal way:

user = User(name='Alice', age=30)
user.age = 31

user.is_dirty()        # True
user.changed_fields()  # {'age': 31}
user.reset_snapshot()  # call after persisting to clear the dirty state

id is declared automatically on every Model (id: RecordID | None = None) — None means the record hasn't been created yet.

Connect and run queries

from surrealdb import AsyncWsSurrealConnection
from surorm import Session

async def main():
    async with AsyncWsSurrealConnection('ws://localhost:8000/rpc') as connection:
        await connection.signin({'username': 'root', 'password': 'root'})
        await connection.use('my_namespace', 'my_database')

        session = Session(connection)

        result = await session.execute(
            Select(User.name, User.age).from_(User).where(User.age > 18)
        )
        users = result.all()  # -> list[User]

Session.execute() returns a Result, which exposes:

• .all() — every row, deserialized into the model class if one was inferred from the statement
• .first() — the first row, or None
• .dicts() — raw rows as plain dicts

Query builder

Statements are frozen dataclasses — every method call returns a new statement, leaving the original untouched:

from surorm.statements import Select, Create, Update, Delete

Select(User.name, User.age).from_(User).where(User.age > 18).sql()
# 'SELECT name, age FROM user WHERE age > 18'

Create(User).content({'name': 'Alice', 'age': 30}).sql()
# 'CREATE user CONTENT { name: "Alice", age: 30 }'

Update(User).set(age=31).where(User.id == 'user:alice').sql()
# 'UPDATE user SET age = 31 WHERE id = user:alice'

Delete(User).where(User.age < 18).sql()
# 'DELETE user WHERE age < 18'

.where() is additive — each call appends a condition (joined with AND), it never replaces the existing clause:

Select(User.name).from_(User).where(User.age > 18).where(User.name == 'Alice').sql()
# 'SELECT name FROM user WHERE age > 18 AND name = "Alice"'

Conditions compose with &, |, and ~:

Select('*').from_(User).where((User.age > 18) & (User.name != 'Alice')).sql()

Other builders follow the same pattern: .limit(), .start(), .order_by(), .fetch(), .merge(), .return_('after'). By default no RETURN clause is emitted — SurrealDB's own default (the mutated record) applies unless you opt in explicitly.

Repository

For straightforward CRUD, wrap a model in a Repository:

from surorm import Repository

class UserRepository(Repository[User]):
    pass

repo = UserRepository(session)

user = await repo.save(User(name='Alice', age=30))  # CREATE (id is None)
user = await repo.update(user, age=31)               # UPDATE
users = await repo.list_(age=31)                      # SELECT * WHERE age = 31
await repo.delete(user)                               # DELETE

save() dispatches automatically: id is None issues a CREATE, otherwise it diffs changed_fields() and issues an UPDATE with only the changed fields.

Relations and graph traversal

SurrealDB relations are edges: separate records that live on their own table and link two other records together with RELATE. surorm supports them in three ways — declaring the edge table, building/executing RELATE statements, and storing direct references with RecordLink.

Declaring an edge table

Subclass Relation instead of Model. in_ and out document the endpoint types; extra fields become properties on the edge itself:

from surorm import Relation
from surorm.data_model import Int

class Likes(Relation):
    __table__ = 'likes'
    in_: User
    out: Post
    rating: Int

Optionally define the table's schema (FROM/TO constrain which tables the edge can connect):

from surorm.statements import DefineTable

DefineTable(Likes).type('relation', from_='user', to='post').sql()
# 'DEFINE TABLE likes SCHEMALESS TYPE RELATION FROM user TO post'

Creating a relation

Build a RELATE statement directly with Relate:

from surorm.statements import Relate

Relate('likes').from_(user.id).to(post.id).sql()
# 'RELATE user:alice->likes->post:1'

Relate('likes').from_(user.id).to(post.id).set(rating=5).return_('after').sql()
# 'RELATE user:alice->likes->post:1 SET rating = 5 RETURN AFTER'

.from_() and .to() each accept multiple records, which relates every source to every target:

Relate('likes').from_(alice.id, bob.id).to(post.id).sql()
# 'RELATE [user:alice,user:bob]->likes->post:1'

.set(**fields) and .content(dict) are mutually exclusive — whichever is called last wins.

Or use Repository.relate(), which takes model instances instead of raw IDs and executes the statement immediately:

repo = UserRepository(session)
await repo.relate(user, Likes, post, rating=5)
# RELATE user:alice->likes->post:1 SET rating = 5 RETURN AFTER

Storing a direct reference with RecordLink

For a plain "points to one other record" field (as opposed to a many-to-many edge), use RecordLink[T] on a regular Model:

from surorm.data_model import RecordLink, String

class Post(Model):
    __table__ = 'post'
    title: String
    author: RecordLink[User]

author holds a RecordID at rest. Add .fetch() to a Select to have SurrealDB resolve it into the full User record instead:

Select('*').from_(Post).fetch(Post.author).sql()
# 'SELECT * FROM post FETCH author'

Schema definitions

from surorm.statements import DefineTable, DefineField, DefineIndex

DefineTable(User).schemafull(True).sql()
# 'DEFINE TABLE user SCHEMAFULL TYPE NORMAL'

DefineField('name', String).on(User).sql()
# 'DEFINE FIELD name ON TABLE user TYPE string'

DefineIndex('user_name_idx').on('user').columns('name').unique().sql()
# 'DEFINE INDEX user_name_idx ON TABLE user COLUMNS name UNIQUE'

Computed fields

A field can be derived from its siblings instead of being assigned directly, via Field(computed=..., computed_in=...). The computed lambda receives the model class and builds an expression from its other fields (cls.field_name); computed_in picks where that expression is evaluated:

from surorm import Field
from surorm.data_model import String, Option

class User(Model):
    __table__ = 'user'
    first_name: String
    last_name: String
    full_name: Option[String] = Field(
        computed=lambda cls: cls.first_name + ' ' + cls.last_name, computed_in='orm'
    )

• computed_in='orm' — no database column. The expression is spliced into the SELECT list at query time, so SELECT * auto-expands to include it:

  Select('*').from_(User).sql()
  # 'SELECT *, first_name + " " + last_name as full_name FROM user'
  

• computed_in='surreal' — a real, server-maintained column. Restate the same expression in a migration with DefineField(...).value(...) (migrations are hand-authored in this repo, so nothing derives the VALUE clause automatically from the field definition):

  from surorm.data_model import Float
  from surorm.operators import Multiply
  from surorm.statements import DefineField
  
  class OrderItem(Model):
      __table__ = 'order_item'
      price: Float
      quantity: Float
      line_total: Option[Float] = Field(
          computed=lambda cls: Multiply(cls.price, cls.quantity), computed_in='surreal'
      )
  
  DefineField('line_total', Float).on(OrderItem).value(Multiply(OrderItem.price, OrderItem.quantity)).sql()
  # 'DEFINE FIELD line_total ON TABLE order_item TYPE float VALUE price * quantity'
  

Class-level access renders context-sensitively — aliased in a top-level SELECT list, bare everywhere else — so a computed field can be used in .where() / .order_by() without referencing a not-yet-projected alias:

Select(User.name).from_(User).where(User.full_name == 'Alice Smith').sql()
# "SELECT name FROM user WHERE first_name + \" \" + last_name = \"Alice Smith\""

Computed fields are always declared as Option[T], are read-only on instances (assigning raises AttributeError), and are excluded from dirty tracking and changed_fields(). They can reference plain fields and other computed fields, with one rule: a surreal-computed field cannot reference an orm-computed one, since the latter has no backing column to read from at the database level.

Embedded documents

Nested, inline documents (no table, no id, no dirty tracking) use EmbeddedModel:

from surorm.orm import EmbeddedModel

class Address(EmbeddedModel):
    city: String
    zip: String

class Customer(Model):
    __table__ = 'customer'
    name: String
    address: Address

Development

poetry install
poetry run pytest
poetry run ruff check .