API Reference

This section contains the complete API reference for Tapestry, automatically generated from the source code.

Core Classes

Base

The foundational class for all Tapestry models.

class Base

Bases: BaseModel

Base class for creating SurrealDB table models with Pydantic.

The Base class provides the foundation for defining SurrealDB tables as Python classes using Pydantic models. All table models should inherit from this class, which handles automatic registration, schema generation, and serialization/deserialization.

id

The SurrealDB record ID. Automatically assigned when records are created in the database. Can be None for new records.

Type:

RecordID | None

Class Attributes:

_registry: Tuple of all registered tables in the system _tokenizers: Set of tokenizer definitions for full-text search _to_create: Queue for chaining object creations child_classes: Dictionary mapping table names to their model classes

Example

>>> from tapestry import Base
>>> from datetime import date
>>>
>>> class Person(Base):
...     first_name: str
...     last_name: str
...     date_of_birth: date
...
>>> # Person is automatically registered and can generate SurrealDB schema
>>> schema = Base.generate_schema()

Notes

• Subclasses are automatically registered upon definition

• Field types are automatically mapped to SurrealDB types

• Supports relationships through Reference fields

• Handles enum serialization automatically

• Provides full-text search capabilities with Text fields

Parameters:

data (Any)

id: RecordID | None

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'use_enum_values': False, 'validate_assignment': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

child_classes: ClassVar[dict[str, type[Base]]] = {}

classmethod validate_record_id(v)

Wrap a classmethod, staticmethod, property or unbound function and act as a descriptor that allows us to detect decorated items from the class’ attributes.

This class’ __get__ returns the wrapped item’s __get__ result, which makes it transparent for classmethods and staticmethods.

wrapped

The decorator that has to be wrapped.

decorator_info

The decorator info.

shim

A wrapper function to wrap V1 style function.

Parameters:

v (Any)

Return type:

RecordID | None

classmethod validate_enums(v, info)

Universal validator for enum fields and record references.

Automatically handles: - Converting enum names (strings) to enum instances - Validating enum values - Processing RecordID references

Parameters:

• v (Any) – The value to validate

• info (ValidationInfo) – Validation context with field information

Return type:

Any

Returns:

The validated/converted value

classmethod registered_tables()

Get the names of all registered tables.

Returns:

Set of table names that have been registered

Return type:

set[str]

Example

>>> tables = Base.registered_tables()
>>> print(tables)
{'person', 'entity', 'role', ...}

classmethod add_table(table, child_class)

Register a new table in the system registry.

Parameters:

• table (Table) – The Table definition to register

• child_class (type[Base]) – The model class associated with the table

Notes

This is called automatically when subclasses are defined. Users typically don’t need to call this directly.

classmethod registered_models()

Get all registered table definitions.

Returns:

List of Table objects that have been registered

Return type:

list[Table]

Example

>>> models = Base.registered_models()
>>> for model in models:
...     print(f"Table: {model.name}")

classmethod generate_schema()

Generate complete SurrealQL schema for all registered tables.

Creates the SQL statements needed to define all tables, fields, indexes, and tokenizers in SurrealDB. This should be executed when setting up a new database or updating the schema.

Returns:

Complete SurrealQL schema definition

Return type:

str

Example

>>> async with AsyncSurreal(url) as db:
...     await db.signin({"username": "root", "password": "root"})
...     await db.use("mydb", "myns")
...     schema = Base.generate_schema()
...     await db.query(schema)

Notes

• Includes table definitions with SCHEMAFULL

• Defines all fields with proper types

• Sets up full-text search indexes

• Configures tokenizers for text analysis

• Creates relationship constraints

classmethod deserialize_record(data)

Deserialize a SurrealDB record into a Pydantic model instance.

Automatically converts SurrealDB records to the appropriate model class based on the record’s table name.

Parameters:

data (dict) – Dictionary containing the record data from SurrealDB

Return type:

Any

Returns:

An instance of the appropriate model class, or the original data if no matching model is found

Notes

• Handles edge records by converting in and out to in_ and out_

• Automatically determines the model class from the record ID

• Validates data using Pydantic validation

classmethod deserialize_response(response)

Deserialize a complete SurrealDB response.

Recursively processes responses to convert all records to their appropriate model instances.

Parameters:

response (Any) – The response from SurrealDB (can be list, dict, or primitive)

Return type:

Any

Returns:

The deserialized response with records converted to model instances

Example

>>> result = await db.select("person")
>>> people = Base.deserialize_response(result)
>>> # people is now a list of Person instances

db_dump()

Serialize the model instance for database insertion/update.

Prepares the model data for sending to SurrealDB by: - Removing the ID field (handled separately by SurrealDB) - Converting Python types to SurrealDB-compatible formats - Serializing nested objects appropriately

Returns:

Dictionary ready for database operations

Return type:

dict[str, Any]

Example

>>> person = Person(first_name="John", last_name="Doe")
>>> data = person.db_dump()
>>> await db.create("person", data)

Node

Used to define entity models (tables) in your database.

class Node

Bases: Base

Base class for SurrealDB node (non-relation) tables.

Node represents standard tables in SurrealDB that are not relationships. All regular tables should inherit from this class to gain database operation methods like insert, create, and save.

Class Attributes:

_is_relation: Always False for Node tables

Example

>>> from tapestry import Node
>>> from datetime import date
>>>
>>> class Person(Node):
...     first_name: str
...     last_name: str
...     date_of_birth: date
...
>>> # Create instances
>>> person = Person(
...     first_name="John",
...     last_name="Doe",
...     date_of_birth=date(1990, 1, 1)
... )
...
>>> # Insert into database
>>> async with db_session as db:
...     await person.create(db)

Notes

• Provides CRUD operations for non-relation tables

• Automatically handles ID assignment from database

• Supports batch inserts for efficiency

• Works with SurrealDB’s SCHEMAFULL tables

Parameters:

data (Any)

async classmethod insert(db, others)

Batch insert multiple records into the database.

Efficiently inserts multiple instances of the model into SurrealDB in a single operation. The original instances are updated with their assigned IDs from the database.

Parameters:

• db (AsyncWsSurrealConnection | AsyncHttpSurrealConnection) – Active SurrealDB connection (WebSocket or HTTP)

• others (Iterable[Self]) – Iterable of model instances to insert

Returns:

The same instances with IDs assigned

Return type:

list[Self]

Example

>>> people = [
...     Person(first_name="John", last_name="Doe"),
...     Person(first_name="Jane", last_name="Smith"),
... ]
>>> inserted = await Person.insert(db, people)
>>> # All instances now have IDs assigned
>>> for person in inserted:
...     print(person.id)

Notes

• More efficient than multiple create() calls

• Original instances are modified in place with IDs

• All records are inserted in a single database transaction

async create(db)

Create a single record in the database.

Inserts this instance into SurrealDB and updates it with the assigned record ID.

Parameters:

db (AsyncWsSurrealConnection | AsyncHttpSurrealConnection) – Active SurrealDB connection (WebSocket or HTTP)

Returns:

The same instance with ID assigned

Return type:

Self

Example

>>> person = Person(
...     first_name="John",
...     last_name="Doe",
...     date_of_birth=date(1990, 1, 1)
... )
>>> await person.create(db)
>>> print(person.id)  # Now has an ID like person:ulid

Notes

• If the instance already has an ID, it will be used

• Otherwise, SurrealDB generates a new unique ID

• The instance is modified in place

async save(db)

Update an existing record in the database.

Saves the current state of this instance to SurrealDB, replacing the existing record with the same ID.

Parameters:

db (AsyncWsSurrealConnection | AsyncHttpSurrealConnection) – Active SurrealDB connection (WebSocket or HTTP)

Returns:

The same instance, potentially with updated fields

Return type:

Self

Example

>>> # Fetch an existing person
>>> person = await Person.get(db, "person:123")
>>> # Modify fields
>>> person.first_name = "Jane"
>>> # Save changes to database
>>> await person.save(db)

Notes

• Requires the instance to have an ID (from create() or query)

• Performs a full replacement of the record

• Use for updating existing records after modifications

Raises:

AssertionError – If the database operation returns unexpected data

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'use_enum_values': False, 'validate_assignment': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Edge

Used to define relationship models between entities.

class Edge

Bases: Base, ABC

Base class for SurrealDB edge (relation) tables.

Edge represents relationship tables in SurrealDB that connect two nodes. All relationship tables should inherit from this class and define in_ and out_ fields to specify the connected node types.

Class Attributes:

_directed (bool): Whether the relationship is directional (default: True)

Required Fields:

in_: The source node of the relationship out_: The target node of the relationship

Example

>>> from tapestry import Edge, Node
>>> from datetime import date
>>>
>>> class Person(Node):
...     name: str
...
>>> class Role(Node):
...     title: str
...
>>> class BelongsTo(Edge):
...     in_: Person  # Person belongs to Role
...     out_: Role
...     begin_date: date
...     end_date: date | None = None
...
>>> # Create a relationship
>>> person = Person(name="John Doe")
>>> role = Role(title="Manager")
>>> belongs = BelongsTo(
...     in_=person,
...     out_=role,
...     begin_date=date(2020, 1, 1)
... )
>>> await belongs.relate(db)

Notes

• Requires both in_ and out_ fields to be defined

• Automatically creates SurrealDB RELATION tables

• Supports directional and bidirectional relationships

• Can have additional fields beyond in_ and out_

• Use relate() instead of create() for edge records

Parameters:

data (Any)

classmethod __init_subclass__(directed=True, **kwargs)

Configure edge subclasses with relationship properties.

Parameters:

• directed (bool, default: True) – Whether the relationship is directional. If False, creates a bidirectional relationship where in/out order doesn’t matter.

• **kwargs – Additional configuration passed to parent

Raises:

TypeError – If the subclass doesn’t define both in_ and out_ fields

db_dump()

Serialize the edge instance for database insertion.

Converts Python field names to SurrealDB format by renaming in_ to in and out_ to out.

Returns:

Dictionary ready for SurrealDB relation operations

Return type:

dict[str, Any]

async relate(db)

Create a relationship record in the database.

Creates an edge record connecting two nodes in SurrealDB. This is the primary method for creating relationships.

Parameters:

db (AsyncWsSurrealConnection | AsyncHttpSurrealConnection) – Active SurrealDB connection (WebSocket or HTTP)

Returns:

The same instance with ID assigned

Return type:

Self

Example

>>> person = await Person.create(db)
>>> role = await Role.create(db)
>>> belongs = BelongsTo(
...     in_=person,
...     out_=role,
...     begin_date=date.today()
... )
>>> await belongs.relate(db)
>>> print(belongs.id)  # Has an ID like belongs_to:xyz

Raises:

Exception – If called on a non-relation table

Notes

• Both in_ and out_ nodes must exist in the database

• Creates a directed or bidirectional edge based on class configuration

• The edge record gets a unique ID from SurrealDB

async classmethod insert(db, others)

Batch insert multiple edge records into the database.

Efficiently creates multiple relationships in a single operation. All edge instances are updated with their assigned IDs.

Parameters:

• db (AsyncWsSurrealConnection | AsyncHttpSurrealConnection) – Active SurrealDB connection (WebSocket or HTTP)

• others (Iterable[Self]) – Iterable of edge instances to insert

Returns:

The same instances with IDs assigned

Return type:

list[Self]

Example

>>> relationships = [
...     BelongsTo(in_=person1, out_=role1, begin_date=date(2020, 1, 1)),
...     BelongsTo(in_=person2, out_=role2, begin_date=date(2021, 1, 1)),
... ]
>>> inserted = await BelongsTo.insert(db, relationships)
>>> # All relationships now have IDs

Notes

• More efficient than multiple relate() calls

• All edges are inserted in a single transaction

• Original instances are modified with IDs

classmethod where(condition)

Add a WHERE condition to this edge for use in graph traversals.

Creates a conditional edge that can be used in query traversals to filter relationships based on their properties.

Parameters:

condition – A condition expression that filters edge records

Returns:

A wrapped edge class with the condition attached

Return type:

EdgeWithCondition

Example

>>> # Find all people who belong to roles that started after 2020
>>> query = (Q(Person) >>
...          BelongsTo.where(BelongsTo.begin_date > date(2020, 1, 1)) >>
...          Role)
>>> results = await query.execute(db)

Notes

• Used primarily in graph traversal queries

• Conditions are applied during traversal, not at definition

• Can filter based on any edge properties

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'use_enum_values': False, 'validate_assignment': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context, /)

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self (BaseModel) – The BaseModel instance.

• context (Any) – The context.

Return type:

None

Query Building

Query

Build and execute queries against SurrealDB.

Query builder for SurrealQL-like syntax in Python.

This module provides a fluent interface for building SurrealQL queries using Python operators and method chaining. The Q class enables type-safe query construction with support for graph traversals, filtering, and projections.

Example

>>> from tapestry import Q, Node
>>>
>>> class Person(Node):
...     name: str
...     age: int
...
>>> # Simple query with filtering
>>> adults = Q(Person).where(Person.age >= 18)
>>> results = await adults.execute(db)
>>>
>>> # Graph traversal query
>>> friends_of_john = (Q(Person)
...     .where(Person.name == "John")
...     >> FriendOf >> Person)
>>> results = await friends_of_john.execute(db)

class Q

Bases: Generic

Main query builder class for constructing SurrealQL queries.

The Q class provides a fluent interface for building complex database queries with support for filtering, graph traversals, and projections. It uses Python operators and method chaining to create readable and type-safe queries.

Type Parameters:

T: The model type being queried (Node or Edge subclass)

table

The Node or Edge class to query from

_select_fields

List of specific fields to select

_where_clause

WHERE condition for filtering

_traversals

List of graph traversals to perform

_value_only

Whether to return only values without field names

Example

>>> # Basic query - type is inferred as Q[Person]
>>> query = Q(Person)
>>>
>>> # With filtering
>>> adults = Q(Person).where(Person.age >= 18)
>>>
>>> # Select specific fields
>>> names = Q(Person).select("first_name", "last_name")
>>>
>>> # Graph traversal using >> operator
>>> managers = Q(Person) >> BelongsTo >> Role.where(Role.title == "Manager")
>>>
>>> # Execute the query - returns list[Person]
>>> results = await query.execute(db)

Notes

• Uses >> for forward traversal and << for backward traversal

• Automatically deserializes results to model instances

• Supports chaining multiple operations

• Generates valid SurrealQL syntax

• Preserves type information for IDE autocomplete

Parameters:

• table (type[TypeVar(T, bound= Base)])

• _select_fields (list[str], default: <factory>)

• _where_clause (str | None, default: None)

• _traversals (list[Traversal], default: <factory>)

• _value_only (bool, default: False)

table: type[T]

select(*fields)

Specify fields to include in the query results.

Parameters:

*fields (str) – Field names to select from the table

Returns:

The query instance for chaining

Return type:

Q[TypeVar(T, bound= Base)]

Example

>>> query = Q(Person).select("first_name", "last_name", "email")
>>> # Generates: SELECT first_name, last_name, email FROM person;

Notes

• If not called, all fields are selected (SELECT *)

• Can be combined with value() for different output formats

value()

Return only values without field names.

Makes the query return raw values instead of objects with field names. Useful for extracting single values or when field names are not needed.

Returns:

The query instance for chaining

Return type:

Q[TypeVar(T, bound= Base)]

Example

>>> # Get just the names as a list of strings
>>> names = Q(Person).select("name").value()
>>> # Generates: SELECT VALUE name FROM person;
>>>
>>> # Get IDs from a traversal
>>> ids = Q(Person).value() >> FriendOf >> Person
>>> # Generates: SELECT VALUE ->friend_of->person.* FROM person;

Notes

• Changes SELECT to SELECT VALUE in the generated query

• Affects the structure of returned results

where(condition)

Add a WHERE clause to filter query results.

Parameters:

condition (Condition) – A Condition object created using field comparisons

Returns:

The query instance for chaining

Return type:

Q[TypeVar(T, bound= Base)]

Example

>>> # Simple condition
>>> adults = Q(Person).where(Person.age >= 18)
>>>
>>> # Complex condition with AND
>>> query = Q(Person).where(
...     (Person.age >= 18) & (Person.city == "Paris")
... )
>>>
>>> # Using OR
>>> query = Q(Person).where(
...     (Person.role == "admin") | (Person.role == "moderator")
... )
>>>
>>> # Negation with ~
>>> active = Q(Person).where(~(Person.status == "deleted"))

Notes

• Conditions are created using field comparisons (==, !=, <, >, <=, >=)

• Use & for AND, | for OR, ~ for NOT

• Supports full-text search with @ operator

traverse(traversal)

Add a graph traversal to navigate relationships.

Parameters:

traversal (Traversal) – A Traversal object defining the path to follow

Returns:

The query instance for chaining

Return type:

Q[TypeVar(T, bound= Base)]

Example

>>> # Direct traversal object
>>> from tapestry.field import Traversal, Direction
>>> traversal = Traversal(Direction.FORWARD, FriendOf)
>>> query = Q(Person).traverse(traversal)

Notes

• Prefer using >> and << operators for simpler syntax

• This method gives more control over traversal configuration

__rshift__(other)

Forward traversal using >> operator for graph navigation.

Follows relationships in the forward direction (from in_ to out_).

Parameters:

other (type[Edge] | type[Node] | Traversal) – An Edge class, Node class, or Traversal to follow

Returns:

The query instance for continued chaining

Return type:

Q[TypeVar(T, bound= Base)]

Example

>>> # Navigate from Person through BelongsTo to Role
>>> managers = Q(Person) >> BelongsTo >> Role
>>> # Generates: SELECT ->belongs_to->role.* FROM person;
>>>
>>> # With conditions on edges
>>> recent = (Q(Person) >>
...          BelongsTo.where(BelongsTo.begin_date > date(2020, 1, 1)) >>
...          Role)
>>>
>>> # Chain multiple traversals
>>> network = Q(Person) >> FriendOf >> Person >> WorksAt >> Company

Notes

• Forward means following from in_ to out_ of edges

• Can traverse through multiple relationships

• Supports conditional edges with .where()

__lshift__(other)

Backward traversal using << operator for reverse graph navigation.

Follows relationships in the backward direction (from out_ to in_).

Parameters:

other (type[Edge] | type[Node] | Traversal) – An Edge class, Node class, or Traversal to follow

Returns:

The query instance for continued chaining

Return type:

Q[TypeVar(T, bound= Base)]

Example

>>> # Find all people who belong to a specific role
>>> members = Q(Role).where(Role.title == "Manager") << BelongsTo << Person
>>> # Generates: SELECT <-belongs_to<-person.* FROM role WHERE title IS "Manager";
>>>
>>> # Reverse traversal to find who manages a department
>>> managers = Q(Department) << Manages << Person
>>>
>>> # Mix forward and backward
>>> related = Q(Person) >> FriendOf >> Person << WorksAt << Company

Notes

• Backward means following from out_ to in_ of edges

• Useful for finding inverse relationships

• Can be combined with forward traversals

to_surreal()

Convert the query to SurrealQL syntax.

Generates the complete SurrealQL query string that can be executed against a SurrealDB database.

Returns:

The SurrealQL query string

Return type:

str

Example

>>> query = (Q(Person)
...     .where(Person.age >= 18)
...     .select("name", "email"))
>>> print(query.to_surreal())
'SELECT name, email FROM person WHERE age >= 18;'
>>>
>>> traversal = Q(Person) >> FriendOf >> Person
>>> print(traversal.to_surreal())
'SELECT ->friend_of->person.* FROM person;'

Notes

• Always ends with a semicolon

• Generates valid SurrealQL syntax

• Handles all query components (SELECT, FROM, WHERE, traversals)

async execute(con)

Execute the query and return deserialized results.

Runs the generated SurrealQL query against the database and automatically deserializes the results into the appropriate model instances based on the Base registry.

Parameters:

con (AsyncWsSurrealConnection | AsyncHttpSurrealConnection) – Active SurrealDB connection (WebSocket or HTTP)

Returns:

The query results, deserialized to model instances

Return type:

list[TypeVar(T, bound= Base)]

Example

>>> # Get all adults as Person instances
>>> query = Q(Person).where(Person.age >= 18)
>>> people = await query.execute(db)
>>> for person in people:
...     print(f"{person.first_name} {person.last_name}")
>>>
>>> # Get names only
>>> names = await Q(Person).select("name").value().execute(db)
>>> print(names)  # ['John', 'Jane', ...]

Notes

• Automatically converts RecordIDs to model instances

• Handles both single records and lists

• Preserves non-record data types

• Uses the Base.deserialize_response for conversion

Raises:

Exception – If the database query fails

__init__(table, _select_fields=<factory>, _where_clause=None, _traversals=<factory>, _value_only=False)

Parameters:

• table (type[TypeVar(T, bound= Base)])

• _select_fields (list[str], default: <factory>)

• _where_clause (str | None, default: None)

• _traversals (list[Traversal], default: <factory>)

• _value_only (bool, default: False)

Database Engine

Engine

Connection and session management.

class PooledConnection

Bases: object

Immutable wrapper around a preconfigured SurrealDB connection.

Instances of this class expose safe query/data operations while deliberately hiding session-mutating methods such as use(). This ensures pooled connections cannot be reconfigured after engine startup.

The wrapped client is expected to already be authenticated and bound to a fixed namespace/database.

Parameters:

client (AsyncWsSurrealConnection | AsyncHttpSurrealConnection)

__init__(client)

Parameters:

client (AsyncWsSurrealConnection | AsyncHttpSurrealConnection)

async query(query, *args, **kwargs)

Parameters:

query (str)

async query_raw(query, *args, **kwargs)

Parameters:

query (str)

async select(record, *args, **kwargs)

async create(record, data=None, *args, **kwargs)

async insert(record, data=None, *args, **kwargs)

async update(record, data=None, *args, **kwargs)

async merge(record, data, *args, **kwargs)

async patch(record, data, *args, **kwargs)

async delete(record, *args, **kwargs)

async let(key, value, *args, **kwargs)

Parameters:

key (str)

async unset(key, *args, **kwargs)

Parameters:

key (str)

async run(name, *args, **kwargs)

Parameters:

name (str)

async version(*args, **kwargs)

async info(*args, **kwargs)

class Engine

Bases: object

SurrealDB connection pool engine with immutable configuration.

The engine represents a pool of authenticated, preconfigured SurrealDB connections. Each connection is bound once at startup to a fixed namespace/database pair and then exposed through an immutable wrapper that does not allow callers to mutate session configuration.

Configuration fields are frozen after initialization. Runtime lifecycle fields remain internal and mutable.

Example

>>> engine = create_engine(
...     url="ws://localhost:8000/rpc",
...     auth_payload={"username": "root", "password": "root"},
...     namespace="my_ns",
...     database="my_db",
...     pool_size=5,
... )
>>> await engine.start()
>>> async with engine.connect() as db:
...     results = await db.select("person")
>>> await engine.close()

Parameters:

• url (str)

• auth_payload (dict[str, Any])

• namespace (str)

• database (str)

• pool_size (int, default: 4)

• connect_timeout (float, default: 5.0)

url: str

auth_payload: dict[str, Any]

namespace: str

database: str

pool_size: int

connect_timeout: float

async start()

Initialize the connection pool with authenticated, preconfigured clients.

Returns:

The engine instance for chaining.

Return type:

Engine

Raises:

• RuntimeError – If the engine is already started or closed.

• Exception – If unable to create the required number of clients.

async close()

Close all connections and shut down the engine.

Return type:

None

connect()

Acquire a preconfigured immutable connection wrapper from the pool.

Yields:

PooledConnection – A connection wrapper that does not expose session-mutating operations like use().

Raises:

RuntimeError – If the engine is not started or is closed.

acquire()

Acquire a preconfigured immutable connection wrapper from the pool.

Yields:

PooledConnection – A connection wrapper that does not expose session-mutating operations like use().

Raises:

RuntimeError – If the engine is not started or is closed.

async execute(query, *args, retries=1, backoff=0.2, **kwargs)

Execute a query using a pooled connection with automatic retry.

Parameters:

• query (str) – The query string to execute

• *args – Arguments forwarded to the query method

• retries (int, default: 1) – Number of retry attempts for transient errors

• backoff (float, default: 0.2) – Base backoff time in seconds between retries

• **kwargs – Keyword arguments forwarded to the query method

__init__(url, auth_payload, namespace, database, pool_size=4, connect_timeout=5.0)

Parameters:

• url (str)

• auth_payload (dict[str, Any])

• namespace (str)

• database (str)

• pool_size (int, default: 4)

• connect_timeout (float, default: 5.0)

create_engine(url, auth_payload, namespace, database, pool_size=4, connect_timeout=5.0)

Create a connection pool engine for SurrealDB with immutable configuration.

Parameters:

• url (str) – SurrealDB connection URL

• auth_payload (dict[str, Any]) – Authentication credentials

• namespace (str) – Fixed namespace configured once for every pooled connection

• database (str) – Fixed database configured once for every pooled connection

• pool_size (int, default: 4) – Number of connections to maintain in the pool

• connect_timeout (float, default: 5.0) – Timeout for connection attempts in seconds

Returns:

An engine instance with immutable configuration.

Return type:

Engine

create_engine_context(url, auth_payload, namespace, database, pool_size=4, connect_timeout=5.0)

Create a connection pool engine with automatic lifecycle management.

The returned engine has immutable configuration, and all pooled connections are preconfigured once with the provided namespace and database.

Parameters:

• url (str)

• auth_payload (dict[str, Any])

• namespace (str)

• database (str)

• pool_size (int, default: 4)

• connect_timeout (float, default: 5.0)

Return type:

AsyncIterator[Engine]

Field Types

Field

Special field types and definitions.

class Field

Bases: Generic

Represents a field in a SurrealDB model with type-safe comparisons.

Parameters:

• name (str)

• model_class (str)

• field_type (type[TypeVar(T)])

__init__(name, model_class, field_type)

Parameters:

• name (str)

• model_class (str)

• field_type (type[TypeVar(T)])

__matmul__(other)

Fulltext search operator using @ symbol.

Parameters:

other (TypeVar(T))

Return type:

Condition

class ComputedFieldDescriptor

Bases: Field, Generic

Descriptor for computed fields that delegates to Pydantic’s property.

Parameters:

• name (str)

• model_class (str)

• field_type (type[TypeVar(T)])

• original_descriptor (property | None, default: None)

__init__(name, model_class, field_type, original_descriptor=None)

Parameters:

• name (str)

• model_class (str)

• field_type (type[TypeVar(T)])

• original_descriptor (property | None, default: None)

__matmul__(other)

Fulltext search operator using @ symbol.

Parameters:

other (TypeVar(T))

Return type:

Condition

class NestedField

Bases: Field, Generic

Field representing a nested path like role.entity.

Parameters:

• path (str)

• parent_class (str)

• field_type (type[TypeVar(T)])

__init__(path, parent_class, field_type)

Parameters:

• path (str)

• parent_class (str)

• field_type (type[TypeVar(T)])

class NestedFieldDescriptor

Bases: object

Descriptor for nested field access in queries.

Parameters:

• field_name (str)

• parent_class (str)

• nested_class (type)

__init__(field_name, parent_class, nested_class)

Parameters:

• field_name (str)

• parent_class (str)

• nested_class (type)

__eq__(other)

Allow direct comparison on the nested field.

Return type:

Condition

__matmul__(other)

Allow fulltext search on the nested field.

Return type:

Condition

class Direction

Bases: Enum

Graph traversal direction.

FORWARD = '->'

BACKWARD = '<-'

BIDIRECTIONAL = '<->'

class Condition

Bases: object

Represents a query condition.

Parameters:

• left (Any)

• operator (str)

• right (Any)

left: Any

operator: str

right: Any

__and__(other)

Combine conditions with AND.

Parameters:

other (Condition)

Return type:

LogicalCondition

__or__(other)

Combine conditions with OR.

Parameters:

other (Condition)

Return type:

LogicalCondition

__invert__()

Negate the condition.

Return type:

NotCondition

to_surreal()

Convert to SurrealQL syntax.

Return type:

str

__init__(left, operator, right)

Parameters:

• left (Any)

• operator (str)

• right (Any)

class LogicalCondition

Bases: object

Represents a logical combination of conditions.

Parameters:

• left (Condition | LogicalCondition)

• operator (str)

• right (Condition | LogicalCondition)

left: Condition | LogicalCondition

operator: str

right: Condition | LogicalCondition

__and__(other)

Combine with another condition using AND.

Parameters:

other (Condition | LogicalCondition)

Return type:

LogicalCondition

__or__(other)

Combine with another condition using OR.

Parameters:

other (Condition | LogicalCondition)

Return type:

LogicalCondition

to_surreal()

Convert to SurrealQL syntax.

Return type:

str

__init__(left, operator, right)

Parameters:

• left (Condition | LogicalCondition)

• operator (str)

• right (Condition | LogicalCondition)

class NotCondition

Bases: object

Represents a negated condition.

Parameters:

condition (Condition | LogicalCondition)

condition: Condition | LogicalCondition

to_surreal()

Convert to SurrealQL syntax.

Return type:

str

__init__(condition)

Parameters:

condition (Condition | LogicalCondition)

class Traversal

Bases: object

Represents a graph traversal operation.

Parameters:

• direction (Direction)

• target (type[Node] | type[Edge] | str)

• recursion_depth (int, default: 1)

• select_fields (str | None, default: None)

• where_condition (Condition | LogicalCondition | None, default: None)

direction: Direction

target: type[Node] | type[Edge] | str

recursion_depth: int = 1

select_fields: str | None = None

where_condition: Condition | LogicalCondition | None = None

recurse(depth)

Set recursion depth for this traversal.

Parameters:

depth (int)

Return type:

Traversal

select(fields='*')

Select specific fields from the traversal result.

Parameters:

fields (str, default: '*')

Return type:

Traversal

where(condition)

Add a WHERE condition to this traversal.

Parameters:

condition (Condition | LogicalCondition)

Return type:

Traversal

to_surreal()

Convert to SurrealQL syntax.

Return type:

str

__init__(direction, target, recursion_depth=1, select_fields=None, where_condition=None)

Parameters:

• direction (Direction)

• target (type[Node] | type[Edge] | str)

• recursion_depth (int, default: 1)

• select_fields (str | None, default: None)

• where_condition (Condition | LogicalCondition | None, default: None)

Utilities

Utils

Utility functions and helpers.

compatible_type(some)

Parameters:

some (Any)

Return type:

Any

convert_types(some)

Parameters:

some (dict[str, Any])

Return type:

dict[str, Any]

flatten_type(tp, base_type=None)

Parameters:

tp (type[Any])

Return type:

Iterable[type]

replace_type(tp, base_type, replacement)

Parameters:

• tp (type[Any])

• base_type (type[Any])

• replacement (type[Any])

Return type:

type[Any]

Table

Table metadata and schema management.

class Indexed

Bases: object

Opt-in index annotation for model fields.

Use with typing.Annotated to request a database index on a field:

from typing import Annotated
from tapestry import Node, Indexed

class Person(Node):
    ext_id: Annotated[str | None, Indexed(unique=True)] = None
    party: Annotated[str | None, Indexed()] = None

unique

If True, the index enforces uniqueness.

Parameters:

unique (bool, default: False)

unique: bool = False

__init__(unique=False)

Parameters:

unique (bool, default: False)

class Link

Bases: object

Link(_in: type[typing.Any], _out: type[typing.Any], symetric: bool = False)

Parameters:

• _in (type[Any])

• _out (type[Any])

• symetric (bool, default: False)

symetric: bool = False

__init__(_in, _out, symetric=False)

Parameters:

• _in (type[Any])

• _out (type[Any])

• symetric (bool, default: False)

class Reference

Bases: Generic

Type annotation for defining foreign key references to other tables.

Reference fields create relationships between tables in SurrealDB by storing RecordIDs that point to records in other tables. This is similar to foreign keys in traditional databases but leverages SurrealDB’s graph capabilities.

Type Parameters:

H: The Node or Edge class that this field references

Example

>>> from tapestry import Node, Reference
>>>
>>> class Person(Node):
...     name: str
...     email: str
...
>>> class Article(Node):
...     title: str
...     author: Reference[Person]  # References a Person record
...     editor: Reference[Person] | None  # Optional reference
...
>>> # When creating an article
>>> person = await Person(name="John Doe", email="john@example.com").create(db)
>>> article = Article(
...     title="Introduction to SurrealDB",
...     author=person  # Can assign the Person instance directly
... )
>>> await article.create(db)
>>>
>>> # The author field will store the Person's RecordID
>>> print(article.author)  # RecordID(table='person', id='...')

Notes

• Reference fields store RecordIDs, not the full referenced object

• Can reference any Node or Edge subclass

• Supports optional references with Union[Reference[T], None]

• When querying, references can be traversed using graph operators

• Unlike Edge tables, References are unidirectional and don’t create separate relation tables

See also

• Edge: For bidirectional relationships with properties

• Node: Base class for tables that can be referenced

class Table

Bases: object

Table(name: str, base_class: type, model_class: type, fields: tuple[tapestry.field.Field[typing.Any], …], name_to_type: dict[str, type] = <factory>, relation: tapestry.table.Link | None = None, index: dict[str, str] = <factory>, tokenizers: tuple[tapestry.tokenizer.Tokenizer, …] = (), record_link_fields: tuple[str, …] = (), indexed_fields: dict[str, bool] = <factory>)

Parameters:

• name (str)

• base_class (type)

• model_class (type)

• fields (tuple[Field[Any], ...])

• name_to_type (dict[str, type], default: <factory>)

• relation (Link | None, default: None)

• index (dict[str, str], default: <factory>)

• tokenizers (tuple[Tokenizer, ...], default: ())

• record_link_fields (tuple[str, ...], default: ())

• indexed_fields (dict[str, bool], default: <factory>)

name: str

base_class: type

model_class: type

fields: tuple[Field[Any], ...]

name_to_type: dict[str, type]

relation: Link | None = None

index: dict[str, str]

tokenizers: tuple[Tokenizer, ...] = ()

record_link_fields: tuple[str, ...] = ()

indexed_fields: dict[str, bool]

static target_name(tp)

generate_table_sql()

Return type:

str

__init__(name, base_class, model_class, fields, name_to_type=<factory>, relation=None, index=<factory>, tokenizers=(), record_link_fields=(), indexed_fields=<factory>)

Parameters:

• name (str)

• base_class (type)

• model_class (type)

• fields (tuple[Field[Any], ...])

• name_to_type (dict[str, type], default: <factory>)

• relation (Link | None, default: None)

• index (dict[str, str], default: <factory>)

• tokenizers (tuple[Tokenizer, ...], default: ())

• record_link_fields (tuple[str, ...], default: ())

• indexed_fields (dict[str, bool], default: <factory>)

Tokenizer

Text tokenization for full-text search.

class Tokenizer

Bases: ABC

Abstract base class for defining text tokenizers for full-text search.

Tokenizers define how text fields are analyzed and indexed for search operations in SurrealDB. Subclasses should implement the define() method to return the SurrealQL DEFINE ANALYZER statement.

name

The name of the tokenizer/analyzer in SurrealDB

name: str

abstractmethod classmethod define()

Generate the SurrealQL definition for this tokenizer.

Returns:

Complete, idempotent SurrealQL definition for the tokenizer

Return type:

str

class FrenchTokenizer

Bases: Tokenizer

Tokenizer for French text with proper handling of elisions and stemming.

This tokenizer: - Removes French elisions (l’, d’, qu’, etc.) - Applies lowercase transformation - Removes accents (ASCII filter) - Applies French stemming using the Snowball algorithm

Example

>>> class Article(Node):
...     title: Text[FrenchTokenizer]
...     content: Text[FrenchTokenizer]

name: str = 'french_analyzer'

classmethod define()

Generate the French analyzer definition for SurrealDB.

Returns:

Complete idempotent analyzer definition including elision function

Return type:

str

class Text

Bases: str, Generic

A string type that enables full-text search indexing in SurrealDB.

Text fields are automatically indexed for full-text search using the specified tokenizer. The tokenizer defines how the text is analyzed, including language-specific processing like stemming and elision removal.

Type Parameters:

T: A Tokenizer subclass that defines the text analysis strategy

Example

>>> from tapestry import Node, Text
>>> from tapestry.tokenizer import FrenchTokenizer
>>>
>>> class Article(Node):
...     title: Text[FrenchTokenizer]  # Indexed for French search
...     content: Text[FrenchTokenizer]
...     summary: str  # Regular string, not indexed
...
>>> # Search using full-text search operator @
>>> articles = await Q(Article).where(
...     Article.title @ "politique"
... ).execute(db)

Notes

• Inherits from str, so can be used wherever strings are expected

• Automatically creates search indexes in SurrealDB

• Supports language-specific text processing

• Use the @ operator in queries for full-text search

• If no tokenizer is specified, defaults to FrenchTokenizer

classmethod __class_getitem__(key)

Parameterize the Text type with a specific tokenizer.

Parameters:

key – Should be a Tokenizer subclass

Returns:

The Text class (for type checking purposes)

Raises:

TypeError – If key is not a Tokenizer subclass

classmethod __get_pydantic_core_schema__(source_type, handler)

Generate Pydantic core schema for Text fields.

This method tells Pydantic that Text fields should be treated as strings for validation purposes, while maintaining the Text type for ORM functionality.

Parameters:

• source_type (Any) – The source type being processed

• handler (GetCoreSchemaHandler) – Pydantic’s schema generation handler

Returns:

A string schema for Pydantic validation

Return type:

Union[InvalidSchema, AnySchema, NoneSchema, BoolSchema, IntSchema, FloatSchema, DecimalSchema, StringSchema, BytesSchema, DateSchema, TimeSchema, DatetimeSchema, TimedeltaSchema, LiteralSchema, MissingSentinelSchema, EnumSchema, IsInstanceSchema, IsSubclassSchema, CallableSchema, ListSchema, TupleSchema, SetSchema, FrozenSetSchema, GeneratorSchema, DictSchema, AfterValidatorFunctionSchema, BeforeValidatorFunctionSchema, WrapValidatorFunctionSchema, PlainValidatorFunctionSchema, WithDefaultSchema, NullableSchema, UnionSchema, TaggedUnionSchema, ChainSchema, LaxOrStrictSchema, JsonOrPythonSchema, TypedDictSchema, ModelFieldsSchema, ModelSchema, DataclassArgsSchema, DataclassSchema, ArgumentsSchema, ArgumentsV3Schema, CallSchema, CustomErrorSchema, JsonSchema, UrlSchema, MultiHostUrlSchema, DefinitionsSchema, DefinitionReferenceSchema, UuidSchema, ComplexSchema]