AsyncDatabase

class nextorm.async_database.AsyncDatabase

Bases: object

Async counterpart to Database.

All I/O methods are coroutines. Use await db.bind(...) etc.

Parameters

entities:

Explicit list of entity classes to include. When omitted, all registered Entity subclasses are used.

__init__(entities=None)

Parameters:

entities (list[type[Entity]] | None)

Return type:

None

async load_lazy_field(entity, field_name)

Asynchronously load a single lazy field and cache the result on the entity.

This must be used for lazy fields when the entity was loaded via AsyncDatabase; synchronous access raises RuntimeError:

article = await db.aselect(Article).fetch_one()
body = await db.load_lazy_field(article, "body")
# or equivalently:
body = await db.load_lazy_field(article, "body")  # cached on second call

Parameters:

• entity (Entity)

• field_name (str)

Return type:

Any

register(*entity_classes)

Explicitly add entity classes to this database.

Parameters:

entity_classes (type[Entity])

Return type:

None

property entities: dict[str, type[Entity]]

Return {entity_name: entity_class} for all entities in this DB.

async bind(provider, *args, **kwargs)

Bind the database to a named async provider and open a connection.

Parameters

provider:

Registered provider name: "sqlite", "postgres", or "mariadb".

args / kwargs:

Connection arguments forwarded to the async provider (e.g. the database path for aiosqlite, or the DSN for PostgreSQL).

Example:

db = AsyncDatabase(entities=[User])
await db.bind("sqlite", ":memory:")
await db.generate_mapping(create_tables=True)

Parameters:

• provider (str)

• args (Any)

• kwargs (Any)

Return type:

None

async close()

Close the async connection (if open).

Return type:

None

property is_bound: bool

Return True when bind() has been called successfully.

async generate_mapping(*, create_tables=False, validate_relations=True)

Build the internal schema and optionally create database tables.

Async counterpart of generate_mapping().

Parameters

create_tables:

When True, execute CREATE TABLE DDL for every entity in the schema.

validate_relations:

When True, every declared Set[T] relation is validated. See generate_mapping() for details.

Parameters:

• create_tables (bool)

• validate_relations (bool)

Return type:

None

property schema: dict[str, Table]

Return a copy of the current table schema (populated by generate_mapping()).

aselect(entity_class)

Return an AsyncQuerySet for entity_class.

generate_mapping() must have been called first.

Example:

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

Parameters:

entity_class (type[ET])

Return type:

AsyncQuerySet

async asave(entity)

Insert or update entity asynchronously.

Async counterpart of save(). PKs that are None trigger an INSERT; a set PK triggers an UPDATE. The entity’s lifecycle hooks are called in both cases.

Example:

async with db_session:
    u = User(name="alice", age=30)
    await db.asave(u)
    print(u.id)  # assigned by DB

Parameters:

entity (Entity)

Return type:

None

async ainsert(entity)

Always INSERT entity asynchronously, even when its primary key is already set.

Async counterpart of insert().

Parameters:

entity (Entity)

Return type:

None

async adelete_instance(entity)

Delete entity from the database asynchronously.

Async counterpart of delete_instance(). Raises ValueError when the entity has not been saved (PK is None).

Parameters:

entity (Entity)

Return type:

None

async acommit()

Flush pending changes then commit the current database transaction.

Calls aflush() first so all pending inserts/updates are written, then finalises the underlying connection’s transaction (PonyORM-style integrated flush).

Return type:

None

async arollback()

Roll back the current database transaction and clear the session cache.

Discards uncommitted work on the active connection and wipes the identity map of the current db_session().

Return type:

None

async aflush()

Write all pending dirty and new objects in the current session to the DB.

Iterates the current db_session() cache and calls asave() on every object that has been scheduled for INSERT or marked dirty.

When called outside a session this is a no-op.

Return type:

None

get_connection()

Return the raw underlying async connection object.

The type depends on the active provider (e.g. aiosqlite.Connection for async SQLite). Use this as an escape hatch for provider-specific features.

Return type:

Any

async execute(sql, *args)

Execute arbitrary SQL asynchronously and return the number of affected rows.

args are bound as positional parameters:

await db.execute("DELETE FROM session WHERE expires_at < ?", cutoff)

Parameters:

• sql (str)

• args (Any)

Return type:

int

async select_raw(sql, *args)

Execute a raw SELECT asynchronously and return results as dicts.

Column names are taken from the cursor description:

rows = await db.select_raw(
    "SELECT u.name, COUNT(o.id) AS cnt "
    "FROM user u LEFT JOIN order o ON o.user_id = u.id "
    "GROUP BY u.id"
)

Parameters:

• sql (str)

• args (Any)

Return type:

list[dict[str, Any]]

property last_sql: str

The last SQL string sent to the database.

property local_stats: dict[str, QueryStat]

Per-database query statistics since the last clear_local_stats() call.

Returns a snapshot copy keyed by SQL string. Each value is a QueryStat with count, sum_time, min_time, max_time, and avg_time attributes.

clear_local_stats()

Reset per-database query statistics.

Return type:

None

merge_local_stats()

Merge per-database stats into the module-level global_stats.

Call this periodically (e.g. at the end of a request) to accumulate statistics across multiple AsyncDatabase instances.

Return type:

None

class nextorm.async_database.AsyncQuerySet

Bases: Generic

Async counterpart to QuerySet.

All terminal methods (fetch_all, fetch_one, etc.) are coroutines.

__init__(entity_class, table, db, builder)

Parameters:

• entity_class (type[T])

• table (Table)

• db (AsyncDatabase)

• builder (SQLBuilder)

Return type:

None

filter(*conditions)

Async counterpart of filter().

Parameters:

conditions (SqlNode)

Return type:

AsyncQuerySet[T]

order_by(*items)

Async counterpart of order_by().

Parameters:

items (OrderItem)

Return type:

AsyncQuerySet[T]

limit(n)

Async counterpart of limit().

Parameters:

n (int)

Return type:

AsyncQuerySet[T]

offset(n)

Async counterpart of offset().

Parameters:

n (int)

Return type:

AsyncQuerySet[T]

join(table_or_entity, on, *, join_type='INNER', alias=None)

Async counterpart of join().

Parameters:

• table_or_entity (type[Entity] | str)

• on (BinOp)

• join_type (str)

• alias (str | None)

Return type:

AsyncQuerySet[T]

async fetch_all()

Async counterpart of fetch_all().

Return type:

list[T]

async fetch_one()

Async counterpart of fetch_one().

Return type:

T | None

async count()

Async counterpart of count().

Return type:

int

async exists()

Async counterpart of exists().

Return type:

bool

async get()

Async counterpart of get().

Return type:

T | None

async get_or_raise()

Async get_or_raise().

Return type:

T

async delete()

Async counterpart of delete().

Return type:

int

async update(**field_values)

Async counterpart of update().

Parameters:

field_values (Any)

Return type:

int

distinct()

Enable SELECT DISTINCT.

Return type:

AsyncQuerySet[T]

without_distinct()

Disable SELECT DISTINCT (reverses a previous distinct() call).

Return type:

AsyncQuerySet[T]

for_update(*, skip_locked=False)

Append FOR UPDATE [SKIP LOCKED].

Parameters:

skip_locked (bool)

Return type:

AsyncQuerySet[T]

page(pagenum, pagesize=10)

Return a page of results (1-based page numbers).

Parameters:

• pagenum (int)

• pagesize (int)

Return type:

AsyncQuerySet[T]

random(n)

Return n randomly ordered rows.

Parameters:

n (int)

Return type:

AsyncQuerySet[T]

where(predicate)

Narrow results using a lambda predicate (same semantics as QuerySet.where).

Parameters:

predicate (Any)

Return type:

AsyncQuerySet[T]

async sum(attr)

Return SUM(attr) or None when no rows match.

Parameters:

attr (str)

Return type:

Any

async avg(attr)

Return AVG(attr) or None when no rows match.

Parameters:

attr (str)

Return type:

Any

async min(attr)

Return MIN(attr) or None when no rows match.

Parameters:

attr (str)

Return type:

Any

async max(attr)

Return MAX(attr) or None when no rows match.

Parameters:

attr (str)

Return type:

Any

async group_concat(attr, sep=',')

Return the concatenation of all non-NULL values of attr.

Uses GROUP_CONCAT(col, sep) on SQLite / MariaDB and STRING_AGG(col::text, sep) on PostgreSQL. Returns None when no rows match or all values are NULL.

Parameters:

• attr (str)

• sep (str)

Return type:

str | None

get_sql()

Return the SQL string that fetch_all() would execute.

Return type:

str

async ashow(width=120, *, file=None)

Async counterpart of show().

Fetches all rows asynchronously and renders them as a plain-text table to file (default: sys.stdout).

Parameters

width:

Maximum total table width in characters.

file:

Output stream; defaults to sys.stdout.

Parameters:

• width (int)

• file (IO[str] | None)

Return type:

None

async raw(sql, params=None)

Execute sql and map each result row to an entity instance.

Column names in the cursor description are matched to entity fields by name. Columns that don’t match any field are silently ignored.

Parameters:

• sql (str)

• params (list[Any] | None)

Return type:

list[T]

async raw_one(sql, params=None)

Execute sql and return the first mapped entity, or None.

Parameters:

• sql (str)

• params (list[Any] | None)

Return type:

T | None

Public API summary

nextorm.async_database.AsyncDatabase.bind(...)	Bind the database to a named async provider and open a connection.
nextorm.async_database.AsyncDatabase.register(...)	Explicitly add entity classes to this database.
nextorm.async_database.AsyncDatabase.generate_mapping(*)	Build the internal schema and optionally create database tables.
nextorm.async_database.AsyncDatabase.aselect(...)	Return an AsyncQuerySet for entity_class.
nextorm.async_database.AsyncDatabase.asave(entity)	Insert or update entity asynchronously.
nextorm.async_database.AsyncDatabase.ainsert(entity)	Always INSERT entity asynchronously, even when its primary key is already set.
nextorm.async_database.AsyncDatabase.adelete_instance(entity)	Delete entity from the database asynchronously.
nextorm.async_database.AsyncDatabase.acommit()	Flush pending changes then commit the current database transaction.
nextorm.async_database.AsyncDatabase.arollback()	Roll back the current database transaction and clear the session cache.
nextorm.async_database.AsyncDatabase.aflush()	Write all pending dirty and new objects in the current session to the DB.
nextorm.async_database.AsyncDatabase.execute(...)	Execute arbitrary SQL asynchronously and return the number of affected rows.
nextorm.async_database.AsyncDatabase.select_raw(...)	Execute a raw SELECT asynchronously and return results as dicts.
nextorm.async_database.AsyncDatabase.close()	Close the async connection (if open).
nextorm.async_database.AsyncDatabase.schema	Return a copy of the current table schema (populated by generate_mapping()).
nextorm.async_database.AsyncDatabase.is_bound	Return True when bind() has been called successfully.
nextorm.async_database.AsyncDatabase.last_sql	The last SQL string sent to the database.
nextorm.async_database.AsyncDatabase.local_stats	Per-database query statistics since the last clear_local_stats() call.
nextorm.async_database.AsyncDatabase.clear_local_stats()	Reset per-database query statistics.
nextorm.async_database.AsyncDatabase.merge_local_stats()	Merge per-database stats into the module-level global_stats.
nextorm.async_database.AsyncDatabase.get_connection()	Return the raw underlying async connection object.