Entity

class nextorm.entity.Entity

Bases: object

Base class for all NextORM entities.

Subclass Entity and annotate fields with NextORM markers to define a mapped table:

from nextorm.fields import Req, Opt, PK, Single, Set, Local


class User(Entity):
    name: Req[str]
    email: Req[str] = Req[str](unique=True)
    age: Opt[int]


class Post(Entity):
    title: Req[str]
    body: Req[LongStr]
    author: Single[User]
    tags: Set[Tag]

Primary keys — a PK[int] auto-increment column named id is injected automatically when no PK is declared. Override explicitly:

class Event(Entity):
    id: PK[uuid7]  # time-ordered UUID, Python-generated
    title: Req[str]

Table name — defaults to the class name lower-cased. Override:

class BlogPost(Entity):
    _table_ = "blog_post"
    title: Req[str]

Composite primary keys — use PrimaryKey():

class OrderLine(Entity):
    order: Single[Order]
    product: Single[Product]
    quantity: Req[int]
    _pk_ = PrimaryKey("order", "product")

Single-table inheritance (STI) — declare _discriminator_col_ on the parent and _discriminator_ on each child; all share one SQL table:

class Animal(Entity):
    _discriminator_col_ = "kind"
    name: Req[str]


class Dog(Animal):
    _discriminator_ = "dog"
    breed: Opt[str]

id: PK[int][int] | Any

__init__(**kwargs)

Initialise a new (unsaved) entity instance.

Pass field values as keyword arguments. Fields not provided here receive their default value (if one was declared) or None:

u = User(name="alice", email="alice@example.com")

If an active session exists, the new instance is automatically scheduled for INSERT on the next flush call.

Parameters:

kwargs (Any)

Return type:

None

after_load()

Called after an existing entity row is loaded from the database.

Override to initialise Local fields or to compute derived state from persisted values:

class Product(Entity):
    name: Req[str]
    price: Req[float]
    _display: Local[str]

    def after_load(self) -> None:
        self._display = f"{self.name} — ${self.price:.2f}"

Return type:

None

before_insert()

Called immediately before a new entity is written to the database.

Override to set computed fields or perform validation before the first INSERT:

class Article(Entity):
    title: Req[str]
    slug: Req[str]

    def before_insert(self) -> None:
        if not self.slug:
            self.slug = self.title.lower().replace(" ", "-")

Return type:

None

after_insert()

Called after a new entity has been written to the database.

The auto-generated primary key (if any) is available at this point:

def after_insert(self) -> None:
    print(f"Saved with id={self.id}")

Return type:

None

before_update()

Called immediately before a dirty entity is flushed to the database.

Override to update timestamps or enforce invariants before UPDATE:

class Post(Entity):
    updated_at: Req[datetime]

    def before_update(self) -> None:
        self.updated_at = datetime.now()

Return type:

None

after_update()

Called after a modified entity has been successfully written to the database.

Return type:

None

before_delete()

Called immediately before an entity is deleted from the database.

Override to perform cleanup or cascading logic that cannot be expressed in the schema:

def before_delete(self) -> None:
    self._cleanup_files()

Return type:

None

after_delete()

Called after an entity has been successfully 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 field values in a single call.

Each assignment goes through the descriptor, so enum coercion, validation, autostrip, and dirty-tracking all apply:

user.set(name="bob", age=31)
# equivalent to:
user.name = "bob"
user.age = 31

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 through a Database so that the _db_ context attribute is set. Use delete_instance() directly if you want to pass the database explicitly.

post = db.select(Post).filter(Post.id == 42).fetch_one()
post.delete()

Raises RuntimeError when the entity has no attached database context (_db_ not set).

Return type:

None

flush()

Persist this entity’s pending changes to the database immediately.

Saves this specific entity — an INSERT when the entity is new (PK not yet assigned), or an UPDATE for dirty fields. Only this instance is written; other entities in the session are unaffected.

The database is resolved in this order:

1. The _db_ context attribute set by a previous db.save() or db.select(...) call.

2. The mapped Database located via _find_db_for_entity().

Raises RuntimeError when no mapped database can be found.

with db_session:
    order = Order(ref="ORD-001")
    order.flush()  # writes only this order — no full session flush
    print(order.id)  # PK available immediately after flush

Return type:

None

commit()

Persist this entity’s changes and commit the underlying transaction.

Equivalent to calling flush() followed by commit() on the attached database. Commits the entire transaction of the attached connection, which is the same scope db.commit() would commit — only this entity’s changes are guaranteed to have been flushed beforehand.

The database is resolved the same way as flush().

with db_session:
    order = Order(ref="ORD-001")
    order.commit()  # flush + commit — order is durable immediately

Return type:

None

classmethod get(**kwargs)

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

Locates the mapped database automatically via _find_db_for_entity(). Raises RuntimeError if more than one row matches.

user = User.get(email="alice@example.com")
if user is None:
    print("not found")

For more complex filtering use select():

user = User.select().filter(User.age > 18).get()

Parameters:

kwargs (Any)

Return type:

Self | None

classmethod exists(**kwargs)

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

if User.exists(email="alice@example.com"):
    raise ValueError("Email already in use")

Parameters:

kwargs (Any)

Return type:

bool

classmethod select()

Return a QuerySet for this entity.

Locates the mapped database automatically. Chain filter, ordering, and fetch methods on the returned queryset:

active = User.select().filter(User.active == True).fetch_all()
count = User.select().count()
first = User.select().filter(User.age > 18).get()

Return type:

Any

classmethod aselect()

Return an AsyncQuerySet for this entity.

Returns the queryset synchronously — no await needed on this call. await is required when executing the query:

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

Raises RuntimeError if the entity is mapped to a sync Database rather than an AsyncDatabase.

Return type:

Any

async classmethod aget(**kwargs)

Async equivalent of get().

Return the first entity matching all given field values, or None. Raises RuntimeError if more than one row matches.

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 field annotations at class-definition time.

When a class that inherits from Entity is created, EntityMeta scans its annotations for NextORM markers (Req, Opt, PK, Single, Set, Local) and performs the following steps:

1. Collects all annotations from the full MRO so that inherited fields are included.

2. For each field marker annotation it builds a FieldSpec, merging type-specific options from the marker call site, and installs a FieldDescriptor.

3. For each relation marker it builds a RelationSpec and installs a SingleDescriptor or SetDescriptor.

4. If no primary key is declared, injects an auto-increment id: PK[int] column automatically.

5. Registers the class in the global _entity_registry so that Database can discover it during generate_mapping().

The following class-level attributes are set on every Entity subclass:

_fields_dict[str, FieldInfo]

All persistent scalar fields keyed by attribute name.

_relations_dict[str, RelationInfo]

All relation fields keyed by attribute name.

_locals_dict[str, LocalInfo]

All transient Local fields keyed by attribute name, each carrying its resolved LocalSpec (default value, py_check).

_pk_fields_tuple[str, ...]

Attribute names of all primary-key fields (one for scalar PKs, multiple for composite PKs declared with PrimaryKey()).

_pk_field_str | None

The single PK attribute name for entities with a scalar PK; None for composite PKs.

_table_name_str

SQL table name — defaults to the class name lower-cased; override by declaring _table_ = "my_table" in the class body.

_constraints_list[CompositeConstraint]

Non-PK composite unique constraints and indexes declared with composite_key() or composite_index().

Single-table inheritance (STI) is supported by declaring _discriminator_col_ on the parent entity and _discriminator_ on each child:

class Animal(Entity):
    _discriminator_col_ = "kind"
    name: Req[str]


class Dog(Animal):
    _discriminator_ = "dog"
    breed: Opt[str]

id: ClassVar[FieldDescriptor[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 metadata for a single persistent field, stored in cls._fields_.

Read-only after creation by EntityMeta. Available for introspection at runtime:

for name, fi in Product._fields_.items():
    print(name, fi.py_type, fi.spec.nullable)

Attributes

name:

The attribute name as declared in the entity class.

py_type:

The Python type used to store and validate the value (e.g. int, str, Vec). UUID sentinel types (uuid7, uuid4, ulid) are normalised to uuid.UUID or ULID here.

spec:

The FieldSpec containing all column options.

__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 metadata for a relation field, stored in cls._relations_.

Read-only after creation by EntityMeta. Available for introspection at runtime:

for name, ri in Comment._relations_.items():
    print(name, ri.spec.kind, ri.spec.target)

Attributes

name:

The attribute name as declared in the entity class.

spec:

The RelationSpec containing all relation options (kind, target entity, nullable, FK column name, etc.).

__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