Engine API

Connection and execution management. Use these helpers when you need lower-level control than the high-level Database and TableHandle APIs (documented in Table API).

QueryExecutor

Query execution helpers (moltres facade over moltres_core.sql).

Translates moltres_core.exceptions into moltres.utils.exceptions so framework integrations (FastAPI, etc.) keep matching registered handlers.

class moltres.engine.execution.QueryExecutor(connection_manager: ConnectionManager, config: EngineConfig)[source]

Bases: QueryExecutor

Same as moltres_core.sql.execution.QueryExecutor with public Moltres exceptions.

execute(sql: str, params: Dict[str, Any] | None = None, transaction: Any = None) → QueryResult[source]

Execute a non-SELECT SQL statement (INSERT, UPDATE, DELETE, etc.).

Parameters:

sql – The SQL statement to execute
params – Optional parameter dictionary for parameterized queries
transaction – Optional transaction connection to use (if None, uses auto-commit)

Returns:

QueryResult with rowcount of affected rows

Raises:

ExecutionError – If SQL execution fails

execute_many(sql: str, params_list: Sequence[Dict[str, Any]], transaction: Any = None) → QueryResult[source]

Execute a SQL statement multiple times with different parameter sets.

This is more efficient than calling execute() in a loop for batch inserts.

Parameters:

sql – The SQL statement to execute
params_list – Sequence of parameter dictionaries, one per execution
transaction – Optional transaction connection to use (if None, uses auto-commit)

Returns:

QueryResult with total rowcount across all executions

Raises:

ExecutionError – If SQL execution fails

fetch(stmt: str | Any, params: Dict[str, Any] | None = None, connection: Any = None, model: Type[Any] | None = None) → QueryResult[source]

Execute a SELECT query and return results.

Parameters:

stmt – The SQLAlchemy Select statement or SQL string to execute
params – Optional parameter dictionary for parameterized queries (only used with SQL strings)
connection – Optional SQLAlchemy Connection to use. If provided, uses this connection directly instead of creating a new one. Useful for executing within existing transactions. The connection’s lifecycle is not managed by this method.

Returns:

QueryResult containing rows and rowcount

Raises:

ExecutionError – If SQL execution fails

class moltres.engine.execution.QueryResult(rows: 'Optional[ResultRows]', rowcount: 'Optional[int]')[source]

Bases: object

rowcount: int | None

rows: Any | None

moltres.engine.execution.register_performance_hook(event: str, callback: Callable[[str, float, dict[str, Any]], None]) → None[source]

Register a performance monitoring hook.

Parameters:

event – Event type - “query_start” or “query_end”
callback – Callback function that receives (sql, elapsed_time, metadata)

Example

>>> def log_slow_queries(sql: str, elapsed: float, metadata: dict):
...     if elapsed > 1.0:
...         print(f"Slow query ({elapsed:.2f}s): {sql[:100]}")
>>> register_performance_hook("query_end", log_slow_queries)

moltres.engine.execution.unregister_performance_hook(event: str, callback: Callable[[str, float, dict[str, Any]], None]) → None[source]

Unregister a performance monitoring hook.

Parameters:

event – Event type - “query_start” or “query_end”
callback – Callback function to remove

ConnectionManager

Synchronous SQLAlchemy connection helpers (provided by moltres_core.sql).

class moltres.engine.connection.ConnectionManager(config: EngineConfig)[source]

Bases: object

Creates and caches SQLAlchemy engines for Moltres sessions.

property active_transaction: sqlalchemy.engine.Connection | None: Get the active transaction connection if one exists.

begin_transaction(savepoint: bool = False, readonly: bool = False, isolation_level: str | None = None, timeout: float | None = None) → sqlalchemy.engine.Connection[source]

Begin a new transaction and return the connection.

Parameters:

savepoint – If True and a transaction is already active, create a savepoint instead.
readonly – If True, set transaction to read-only mode.
isolation_level – Optional isolation level (READ UNCOMMITTED, READ COMMITTED, REPEATABLE READ, SERIALIZABLE).
timeout – Optional transaction timeout in seconds.

Returns:

Connection that is part of a transaction (not auto-committed)

Raises:

RuntimeError – If savepoint=False and a transaction is already active.
ValueError – If isolation level or readonly is requested but not supported by dialect.

close() → None[source]: Rollback any active transaction and dispose the engine if owned.

commit_transaction(connection: sqlalchemy.engine.Connection) → None[source]

Commit a transaction.

Parameters:: connection – The transaction connection to commit

connect(transaction: sqlalchemy.engine.Connection | None = None) → Iterator[sqlalchemy.engine.Connection][source]

Get a database connection.

Parameters:: transaction – If provided, use this transaction connection instead of creating a new one. This allows operations to share a transaction. If None and an active transaction exists, uses the active transaction.
Yields:: Database connection

create_savepoint(connection: sqlalchemy.engine.Connection, name: str) → sqlalchemy.engine.Connection[source]

Create a savepoint in the current transaction.

Parameters:

connection – The transaction connection
name – Savepoint name

Returns:

The same connection (for compatibility)

Raises:

RuntimeError – If no transaction is active or connection doesn’t match active transaction.

property engine: sqlalchemy.engine.Engine

release_savepoint(connection: sqlalchemy.engine.Connection, name: str) → None[source]

Release a savepoint.

Parameters:

connection – The transaction connection
name – Savepoint name to release

Raises:

RuntimeError – If no transaction is active, connection doesn’t match, or savepoint not found.

rollback_to_savepoint(connection: sqlalchemy.engine.Connection, name: str) → None[source]

Rollback to a specific savepoint.

Parameters:

connection – The transaction connection
name – Savepoint name to rollback to

Raises:

RuntimeError – If no transaction is active, connection doesn’t match, or savepoint not found.

rollback_transaction(connection: sqlalchemy.engine.Connection) → None[source]

Rollback a transaction.

Parameters:: connection – The transaction connection to rollback

property savepoint_stack: list[str]: Get the current savepoint stack.

property transaction_metadata: dict[str, object] | None: Get transaction metadata if a transaction is active.