Database

class nextorm.database.Database

Bases: object

Owns a provider binding, a persistent connection, and entity schema.

The database keeps a single open connection so that in-memory SQLite databases (":memory:") remain accessible across multiple operations.

Call close() (or use the database as a context manager) to release the connection when done.

Parameters

entities:

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

__init__(entities=None)

Parameters:

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

Return type:

None

register(*entity_classes)

Add one or more entity classes to this database’s scope.

Entities register themselves globally at import time; call this method to opt specific classes into a database that was constructed with an explicit entities=[...] list, or to add entities after construction.

Calling register() with an entity that is already registered is a no-op. Each argument must be an Entity subclass — passing anything else raises TypeError.

Example:

db = Database(entities=[User])
db.register(Post, Comment)  # add more after the fact
db.bind("sqlite", ":memory:")
db.generate_mapping(create_tables=True)

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.

bind(provider, *args, pool_min=0, pool_max=1, pool_timeout=30.0, **kwargs)

Bind the database to a named provider.

Parameters

provider:

Registered provider name ("sqlite", "postgres", "mariadb").

*args / **kwargs:

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

pool_min:

Minimum (pre-created) pool connections. 0 means no connections are opened until the first query.

pool_max:

Maximum pool size. 1 (the default) keeps the legacy single-persistent-connection behaviour. Values > 1 enable true pooling — each operation checks out a connection from the pool and returns it afterwards.

pool_timeout:

Seconds to wait for a free connection before PoolTimeoutError is raised.

Parameters:

• provider (str)

• args (Any)

• pool_min (int)

• pool_max (int)

• pool_timeout (float)

• kwargs (Any)

Return type:

None

unbind()

Close the connection and clear the provider binding.

Return type:

None

property is_bound: bool

Return True when bind() has been called successfully.

close()

Close the persistent connection (if open) or all pool connections.

Return type:

None

commit()

Flush pending changes then commit the current database transaction.

Calls flush() first so all pending inserts/updates are written, then finalises the underlying connection’s transaction. This mirrors PonyORM’s approach where commit() integrates the flush step internally — callers never need to flush manually before committing.

Return type:

None

rollback()

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(). When called outside a session the cache-clear step is a no-op.

Return type:

None

flush()

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

Iterates the current db_session() cache and calls save() on every object that has been scheduled for INSERT or marked dirty that belongs to this database. Because NextORM auto-commits per DML statement, each save is independently committed.

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

Return type:

None

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

Build the internal schema and optionally create database tables.

Must be called after bind(). Registers this database instance so that _find_db_for_entity() can locate it.

Parameters

create_tables:

When True, execute CREATE TABLE DDL for every entity in the schema. The connection is kept open so that in-memory SQLite databases (":memory:") remain accessible after the call.

validate_relations:

When True, every declared Set[T] relation is checked to have a matching Single[T] back-reference on the target entity. Ambiguous cases (multiple candidates with no reverse=) raise MappingError.

Example:

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

Parameters:

• create_tables (bool)

• validate_relations (bool)

Return type:

None

property schema: dict[str, Table]

Current schema mapping (empty until generate_mapping() is called).

migrate()

Apply pending schema changes to the live database.

Introspects the current database schema, computes the diff against the entity-derived target schema (built by generate_mapping()), and executes each pending DDL operation (CREATE TABLE, ALTER TABLE … ADD/DROP COLUMN, CREATE/DROP INDEX, etc.).

Returns the list of SQL statements that were executed. An empty list means the database is already up to date.

Raises RuntimeError if the database is not bound to a provider or if generate_mapping() has not been called yet.

Return type:

list[str]

get_ddl()

DDL statements from the last generate_mapping() call.

Return type:

list[str]

select(entity_class)

Return a QuerySet for entity_class.

generate_mapping() must have been called first.

Parameters:

entity_class (type[T])

Return type:

QuerySet[T]

save(entity)

Insert or update entity in the database.

• All PK columns are None → INSERT (auto-PK written back when single auto PK).

• At least one PK is set → UPDATE all non-PK columns.

Lifecycle hooks are called around each operation. When a db_session() is active the entity is registered in its identity map after a successful save.

Parameters:

entity (Entity)

Return type:

None

insert(entity)

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

Unlike save(), which uses the PK value to choose between INSERT and UPDATE, this method always performs an INSERT. Use it for:

• Entities with user-assigned (non-auto) primary keys.

• Data loading / migration where you know the row does not yet exist.

• Inserting a copy of an entity with a freshly assigned identity.

When the PK is auto-generated (FieldSpec(primary_key=True, auto=True)):

• PK is None → the database generates one, which is written back.

• PK is a concrete value → that value is passed in the INSERT, overriding the auto-increment counter (useful for data migration).

Lifecycle hooks before_insert / after_insert are called.

Parameters:

entity (Entity)

Return type:

None

delete_instance(entity)

Delete entity from the database and clear its primary key.

Raises ValueError when the entity has not been saved (PK is None).

Parameters:

entity (Entity)

Return type:

None

get_connection()

Return the raw underlying DBAPI connection.

Use this as an escape hatch when you need provider-specific features not covered by the NextORM API. The returned type depends on the active provider (e.g. sqlite3.Connection for SQLite).

Raises RuntimeError when the database is not connected.

Return type:

Any

execute(sql, *args)

Execute arbitrary SQL and return the number of affected rows.

args are passed as positional parameters to the DBAPI cursor (i.e. bound as ? or %s placeholders depending on the provider).

This is useful for DDL statements or DML that cannot be expressed via the QuerySet API:

db.execute("CREATE INDEX idx_user_email ON user (email)")
db.execute("DELETE FROM session WHERE expires_at < ?", cutoff)

Parameters:

• sql (str)

• args (Any)

Return type:

int

select_raw(sql, *args)

Execute a raw SELECT and return the results as a list of dicts.

Column names are taken from the cursor description. Use this when you need a join result or aggregate that doesn’t map to a single entity class:

rows = db.select_raw(
    "SELECT u.name, COUNT(o.id) AS order_count "
    'FROM user u LEFT JOIN "order" o ON o.user_id = u.id '
    "GROUP BY u.id"
)
for row in rows:
    print(row["name"], row["order_count"])

Parameters:

• sql (str)

• args (Any)

Return type:

list[dict[str, Any]]

property last_sql: str

The last SQL string sent to the database.

Updated after every _execute, _execute_dml, and _execute_insert call. Empty string if no query has been executed yet. Useful for debugging:

db.select(User).filter(User.age > 18).fetch_all()
print(db.last_sql)

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 Database instances.

Return type:

None

Public API summary

nextorm.database.Database.bind(provider, *args)	Bind the database to a named provider.
nextorm.database.Database.unbind()	Close the connection and clear the provider binding.
nextorm.database.Database.register(...)	Add one or more entity classes to this database's scope.
nextorm.database.Database.generate_mapping(*)	Build the internal schema and optionally create database tables.
nextorm.database.Database.select(entity_class)	Return a QuerySet for entity_class.
nextorm.database.Database.save(entity)	Insert or update entity in the database.
nextorm.database.Database.insert(entity)	Always INSERT entity, even when its primary key is already set.
nextorm.database.Database.delete_instance(entity)	Delete entity from the database and clear its primary key.
nextorm.database.Database.commit()	Flush pending changes then commit the current database transaction.
nextorm.database.Database.rollback()	Roll back the current database transaction and clear the session cache.
nextorm.database.Database.flush()	Write all pending dirty and new objects in the current session to the DB.
nextorm.database.Database.execute(sql, *args)	Execute arbitrary SQL and return the number of affected rows.
nextorm.database.Database.select_raw(sql, *args)	Execute a raw SELECT and return the results as a list of dicts.
nextorm.database.Database.get_ddl()	DDL statements from the last generate_mapping() call.
nextorm.database.Database.migrate()	Apply pending schema changes to the live database.
nextorm.database.Database.close()	Close the persistent connection (if open) or all pool connections.
nextorm.database.Database.schema	Current schema mapping (empty until generate_mapping() is called).
nextorm.database.Database.is_bound	Return True when bind() has been called successfully.
nextorm.database.Database.last_sql	The last SQL string sent to the database.
nextorm.database.Database.local_stats	Per-database query statistics since the last clear_local_stats() call.
nextorm.database.Database.clear_local_stats()	Reset per-database query statistics.
nextorm.database.Database.merge_local_stats()	Merge per-database stats into the module-level global_stats.
nextorm.database.Database.get_connection()	Return the raw underlying DBAPI connection.