Session

nextorm.session.db_session = DBSessionManager instance

Unified sync/async context manager and decorator for database sessions.

The same db_session object can be used in four ways:

with db_session:  # sync context manager (no-arg)
    ...

with db_session():  # sync context manager (call form)
    ...


@db_session  # sync decorator
def f(): ...


@db_session  # async decorator (detected at call time)
async def f(): ...

Parametrised form:

with db_session(sql_debug=True):
    ...


@db_session(retry=3)
def f(): ...

Nesting is supported — only the outermost context actually clears the cache on exit.

Parameters

retry:

Number of additional attempts to make after the first one fails with TransactionError (e.g. a deadlock or serialisation failure). retry=3 means up to 4 total attempts. Applies only when used as a decorator.

sql_debug:

When True, enables SQL debug logging for the duration of the session and restores the previous state on exit. Equivalent to wrapping the body with sql_debugging.

serializable:

Hint that the session should use SERIALIZABLE transaction isolation. Recorded on the session cache; enforcement is the responsibility of the provider / middleware layer.

immediate:

Hint that the session should lock immediately (SQLite BEGIN IMMEDIATE). Recorded on the session cache; enforcement is the responsibility of the provider / middleware layer.

nextorm.session.async_db_session = alias for db_session

Convenience alias — behaves identically to db_session but signals async intent to readers of the calling code.

class nextorm.session.DBSessionManager

Bases: object

Unified sync/async context manager and decorator for database sessions.

The same db_session object can be used in four ways:

with db_session:  # sync context manager (no-arg)
    ...

with db_session():  # sync context manager (call form)
    ...


@db_session  # sync decorator
def f(): ...


@db_session  # async decorator (detected at call time)
async def f(): ...

Parametrised form:

with db_session(sql_debug=True):
    ...


@db_session(retry=3)
def f(): ...

Nesting is supported — only the outermost context actually clears the cache on exit.

Parameters

retry:

Number of additional attempts to make after the first one fails with TransactionError (e.g. a deadlock or serialisation failure). retry=3 means up to 4 total attempts. Applies only when used as a decorator.

sql_debug:

When True, enables SQL debug logging for the duration of the session and restores the previous state on exit. Equivalent to wrapping the body with sql_debugging.

serializable:

Hint that the session should use SERIALIZABLE transaction isolation. Recorded on the session cache; enforcement is the responsibility of the provider / middleware layer.

immediate:

Hint that the session should lock immediately (SQLite BEGIN IMMEDIATE). Recorded on the session cache; enforcement is the responsibility of the provider / middleware layer.

__init__(*, retry=0, sql_debug=False, show_values=False, serializable=False, immediate=False, optimistic=True, strict=False, allowed_exceptions=None, retry_exceptions=None)

Parameters:

• retry (int)

• sql_debug (bool)

• show_values (bool)

• serializable (bool)

• immediate (bool)

• optimistic (bool)

• strict (bool)

• allowed_exceptions (list[type[BaseException]] | None)

• retry_exceptions (list[type[BaseException]] | None)

Return type:

None

__enter__()

Return type:

SessionCache

__exit__(exc_type, exc_val, exc_tb)

Parameters:

• exc_type (Any)

• exc_val (Any)

• exc_tb (Any)

Return type:

bool | None

__call__(func: Callable[_P, _T], /) → Callable[_P, _T]

__call__(func: None = None, /, *, retry: int | None = None, sql_debug: bool | None = None, show_values: bool | None = None, serializable: bool | None = None, immediate: bool | None = None, optimistic: bool | None = None, strict: bool | None = None, allowed_exceptions: list[type[BaseException]] | None = None, retry_exceptions: list[type[BaseException]] | None = None) → Self

Called as db_session(...) to create a parametrised manager, or as a decorator.

When called without a callable argument (e.g. db_session(retry=3)), returns a new DBSessionManager configured with the given parameters. When called with a callable (@db_session bare decorator), wraps it.

Parameters:

• func (Callable[..., Any] | None)

• retry (int | None)

• sql_debug (bool | None)

• show_values (bool | None)

• serializable (bool | None)

• immediate (bool | None)

• optimistic (bool | None)

• strict (bool | None)

• allowed_exceptions (list[type[BaseException]] | None)

• retry_exceptions (list[type[BaseException]] | None)

Return type:

Callable[…, Any] | Self

async __aenter__()

Return type:

SessionCache

async __aexit__(exc_type, exc_val, exc_tb)

Parameters:

• exc_type (Any)

• exc_val (Any)

• exc_tb (Any)

Return type:

bool | None

as_context()

Explicit generator-based context (useful in pytest with with blocks).

Return type:

Generator[SessionCache, None, None]

as_async_context()

Explicit async generator-based context.

Return type:

AsyncGenerator[SessionCache, None]

property depth: int

Current nesting depth (0 = not inside any session).

static current()

Return the innermost active SessionCache, or None.

Return type:

SessionCache | None

class nextorm.session.SessionCache

Bases: object

Identity map and dirty-object tracker for a single database session.

The cache is keyed by (entity_class, primary_key_value) which ensures that at most one Python object exists per database row inside a session.

Attributes

_objects:

The identity map: {(entity_cls, pk): entity_instance}.

_dirty:

Set of entity instances that have been modified and need flushing.

_to_save:

Instances scheduled for INSERT (not yet persisted).

_modified_collections:

Tracks M2M collections that have pending additions/removals.

__init__()

Return type:

None

serializable: bool

Set by DBSessionManager when serializable=True was requested.

immediate: bool

Set by DBSessionManager when immediate=True was requested.

optimistic: bool

Set to False by DBSessionManager when optimistic=False was requested. When True (the default), save() adds AND col = orig_val clauses for the columns the caller has READ since load, implementing Pony-style per-field optimistic concurrency detection.

strict: bool

When True, entity objects should not be accessed after the session ends. Currently the cache is always cleared; strict=False retains the same behaviour — no cross-session lazy-load is supported yet.

get(key)

Return the cached instance for key, or None.

Parameters:

key (tuple[type[Entity], Any])

Return type:

Entity | None

put(entity, pk)

Add entity to the identity map under its (type, pk) key.

Parameters:

• entity (Entity)

• pk (Any)

Return type:

None

remove(entity, pk)

Remove entity from the identity map.

Parameters:

• entity (Entity)

• pk (Any)

Return type:

None

clear()

Wipe all cached state (called on session commit / rollback).

Return type:

None

mark_dirty(entity)

Mark entity as modified so it will be flushed on commit.

Parameters:

entity (Entity)

Return type:

None

unmark_dirty(entity)

Remove entity from the dirty set (e.g. after successful flush).

Parameters:

entity (Entity)

Return type:

None

property dirty_objects: frozenset[Entity]

Snapshot of currently-dirty entity instances.

schedule_save(entity)

Enqueue entity for INSERT on the next commit.

Parameters:

entity (Entity)

Return type:

None

unschedule_save(entity)

Remove entity from the INSERT queue (called after a successful save).

Parameters:

entity (Entity)

Return type:

None

property objects_to_save: list[Entity]

Ordered list of entities pending INSERT.

track_collection_change(owner, attr, action, related)

Record that related was added to or removed from owner.*attr*.

Parameters

action:

"add" or "remove".

Parameters:

• owner (Entity)

• attr (str)

• action (str)

• related (Entity)

Return type:

None

property modified_collections: dict[tuple[Entity, str], dict[str, list[Entity]]]

Snapshot of pending M2M collection mutations.

Module-level helpers

nextorm.__init__.flush()

Flush all pending changes to all databases in the current session.

Iterates the dirty set and new-objects queue of the active db_session(), groups them by Database, and calls flush() on each.

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

Return type:

None

nextorm.__init__.commit()

Commit all sync databases that have entities in the current session.

Mirrors PonyORM’s staged-commit strategy:

1. Flush all databases first. On any flush error every database is rolled back and the exception is re-raised.

2. Commit the primary database (first in the list) first. On failure all secondary databases are rolled back and a CommitException is raised.

3. Commit the remaining databases. On any failure a PartialCommitException is raised (the primary commit is already durable at this point).

When called outside a session this is a no-op. For async databases use await db.acommit() directly.

Return type:

None

nextorm.__init__.rollback()

Roll back all databases that have entities in the current session.

Iterates the identity map of the active db_session(), collects the unique Database objects referenced by cached entities, calls rollback() on each, and then clears the session cache.

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

Return type:

None