surorm 0.0.12

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, 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

Edge tables are declared with Relation instead of Model:

from surorm import Relation
from surorm.statements import Relate

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

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

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'

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 .