Entity

class nextorm.entity.Entity

Bases: object

Base class for all NextORM entities.

Define fields using type aliases:

class Product(Entity):
    name: Req[str]
    price: Req[float]
    description: Opt[str]

Fields are discovered automatically; no db.Entity coupling needed.

id: PK[int]

__init__(**kwargs)

Parameters:

kwargs (Any)

Return type:

None

after_load()

Called after an existing entity is loaded from the database.

Return type:

None

before_insert()

Called before a new entity is saved to the database for the first time.

Return type:

None

after_insert()

Called after a new entity is saved to the database for the first time.

Return type:

None

before_update()

Called before a modified entity is saved to the database.

Return type:

None

after_update()

Called after a modified entity is saved to the database.

Return type:

None

before_delete()

Called before an entity is deleted from the database.

Return type:

None

after_delete()

Called after an entity is deleted from the database.

Return type:

None

get_pk()

Return the primary key value of this entity instance.

Returns None when the entity has no primary key field or when the PK has not been set yet (e.g. before the first db.save() call).

Example:

u = User(name="alice")
db.save(u)
pk = u.get_pk()  # → 1

Return type:

Any

set(**kwargs)

Bulk-assign attribute values in a single call.

Equivalent to calling setattr(self, k, v) for every provided keyword argument. Dirty-tracking in the active session is updated automatically because each assignment goes through the descriptor:

user.set(name="bob", age=30)

Parameters:

kwargs (Any)

Return type:

None

delete()

Delete this entity from the database using its attached database context.

The entity must have been loaded or saved via a Database (or AsyncDatabase) so that the _db_ context attribute is available. Use delete_instance() directly if you want to pass the database explicitly.

Raises RuntimeError when the entity has no attached database context.

Return type:

None

classmethod get(**kwargs)

Return the first entity matching all given field values, or None.

Raises RuntimeError if more than one row matches. Uses _find_db_for_entity() to locate the mapped database.

Example:

user = User.get(name="alice")

Parameters:

kwargs (Any)

Return type:

Self | None

classmethod exists(**kwargs)

Return True if at least one entity matches all given field values.

Uses _find_db_for_entity() to locate the mapped database.

Example:

if User.exists(name="alice"):
    ...

Parameters:

kwargs (Any)

Return type:

bool

classmethod select()

Return a QuerySet for this entity.

Convenience shortcut for db.select(Entity) that locates the mapped database automatically.

Example:

active_users = User.select().filter(User.active == True).fetch_all()
count = User.select().count()

Return type:

Any

classmethod aselect()

Return an AsyncQuerySet for this entity.

Sync builder — returns an AsyncQuerySet directly (no await needed). Async counterpart of select(). Locates the mapped AsyncDatabase automatically.

Example:

users = await User.aselect().filter(User.active == True).fetch_all()

Return type:

Any

async classmethod aget(**kwargs)

Return the first entity matching all given field values, or None.

Async counterpart of get(). Raises RuntimeError if more than one row matches.

Example:

user = await User.aget(email="alice@example.com")

Parameters:

kwargs (Any)

Return type:

Self | None

to_dict(only=None, exclude=None, *, with_collections=False, with_lazy=False, related_objects=False)

Serialize this entity to a plain Python dictionary.

Parameters

only:

If given, only the listed field names are included. Relations are included only when with_collections or related_objects is also True.

exclude:

Field names to exclude. Applied after only.

with_collections:

When True, Set[T] relation attributes are included as lists of their to_dict() results. Requires the collections to have been prefetched or lazily loaded beforehand.

with_lazy:

When False (the default) lazy fields that have not yet been loaded are omitted from the result. Pass True to include them; unloaded lazy fields are fetched on demand (sync databases only).

related_objects:

When True, loaded Single[T] relation attributes are included as nested to_dict() results. If the related object has not been loaded, the raw FK value ({name}_id) is included instead.

Example:

user.to_dict()
# → {"id": 1, "name": "alice", "age": 30}

user.to_dict(exclude=["id"])
# → {"name": "alice", "age": 30}

user.to_dict(with_collections=True)
# → {"id": 1, ..., "posts": [{"id": 5, "title": "hi", ...}]}

article.to_dict(with_lazy=True)
# → {"id": 1, "title": "hi", "body": "<full text>"}

comment.to_dict(related_objects=True)
# → {"id": 1, "text": "great", "author": {"id": 2, "name": "bob"}}

Parameters:

• only (list[str] | None)

• exclude (list[str] | None)

• with_collections (bool)

• with_lazy (bool)

• related_objects (bool)

Return type:

dict[str, Any]

classmethod select_by_sql(db, sql, params=None)

Execute sql and return all rows mapped to instances of this entity.

Convenience wrapper around db.select(cls).raw(sql, params).

Example:

users = User.select_by_sql(db, "SELECT * FROM user WHERE age > ?", [18])

Parameters:

• db (Database)

• sql (str)

• params (list[Any] | None)

Return type:

list[Self]

classmethod get_by_sql(db, sql, params=None)

Execute sql and return the first row as an entity instance, or None.

Convenience wrapper around db.select(cls).raw_one(sql, params).

Example:

user = User.get_by_sql(db, "SELECT * FROM user WHERE id = ?", [1])

Parameters:

• db (Database)

• sql (str)

• params (list[Any] | None)

Return type:

Self | None

async classmethod aselect_by_sql(db, sql, params=None)

Async equivalent of select_by_sql().

Example:

users = await User.aselect_by_sql(db, "SELECT * FROM user WHERE age > %s", [18])

Parameters:

• db (AsyncDatabase)

• sql (str)

• params (list[Any] | None)

Return type:

list[Self]

async classmethod aget_by_sql(db, sql, params=None)

Async equivalent of get_by_sql().

Example:

user = await User.aget_by_sql(db, "SELECT * FROM user WHERE id = %s", [1])

Parameters:

• db (AsyncDatabase)

• sql (str)

• params (list[Any] | None)

Return type:

Self | None

class nextorm.entity.EntityMeta

Bases: type

Metaclass that processes Annotated field specs at class definition time.

id: PK[int]

static __new__(mcs, name, bases, namespace, **kwargs)

Parameters:

• name (str)

• bases (tuple[type, ...])

• namespace (dict[str, Any])

• kwargs (Any)

Return type:

EntityMeta

class nextorm.entity.FieldInfo

Bases: object

Resolved information about a single persistent field.

__init__(name, py_type, spec)

Parameters:

• name (str)

• py_type (type)

• spec (FieldSpec)

Return type:

None

name

py_type

spec

class nextorm.entity.RelationInfo

Bases: object

Resolved information about a relation field.

__init__(name, spec)

Parameters:

• name (str)

• spec (RelationSpec)

Return type:

None

name

spec

class nextorm.collection.RelatedCollection

Bases: Generic

A lazy, database-backed collection representing one side of a relation.

Instances are created by SetDescriptor on first attribute access. They hold a weak logical reference to the owner entity so that individual collections can be GC’d when not needed.

The collection is not thread-safe. Use separate collection objects in separate threads.

Parameters

owner:

The entity instance that owns this relation attribute.

ri:

The RelationInfo for this relation.

db:

The Database to query. None when the owner entity was not loaded from a database (e.g. a freshly created but unsaved entity).

__init__(owner, ri, db)

Parameters:

• owner (Entity)

• ri (RelationInfo)

• db (Database | None)

Return type:

None

count()

Return the number of items without loading them.

Return type:

int

is_empty()

Return True if the collection has no items.

Return type:

bool

copy()

Return a plain Python set of all loaded items.

Return type:

set[T]

load()

Eagerly load all items and return them as a list.

Return type:

list[T]

select()

Return a QuerySet scoped to this collection.

Return type:

QuerySet[T]

filter(*conditions)

Return a filtered QuerySet for this collection.

Parameters:

conditions (Any)

Return type:

QuerySet[T]

order_by(*items)

Return an ordered QuerySet for this collection.

Parameters:

items (Any)

Return type:

QuerySet[T]

page(pagenum, pagesize=10)

Return a page of this collection’s items (1-based page numbers).

Parameters:

• pagenum (int)

• pagesize (int)

Return type:

QuerySet[T]

random(n)

Return n randomly selected items from this collection.

Parameters:

n (int)

Return type:

QuerySet[T]

add(*items)

Add one or more items to this collection.

For M2M: inserts rows into the join table. For O2M: updates the FK column on each item.

Parameters:

items (T)

Return type:

None

remove(*items)

Remove one or more items from this collection.

For M2M: deletes rows from the join table. For O2M: sets the FK column to NULL (requires nullable FK).

Parameters:

items (T)

Return type:

None

clear()

Remove all items from this collection.

Return type:

None

create(**kwargs)

Create a related entity and immediately link it to this collection.

The entity is saved to the database first (to obtain a primary key), then linked via add(). Returns the newly created entity.

Parameters:

kwargs (Any)

Return type:

T