Native Engine¶
The native engine wraps the Rust execution layer exposed by ormdantic._ormdantic.
runtime_capabilities()
Return the database runtimes compiled into the Rust extension.
Source code in ormdantic/engine.py
def runtime_capabilities() -> dict[str, bool]:
"""Return the database runtimes compiled into the Rust extension."""
if _ormdantic is None or not hasattr(_ormdantic, "runtime_capabilities"):
return {
"sqlite": False,
"postgresql": False,
"mysql": False,
"mariadb": False,
"mssql": False,
"oracle": False,
}
return dict(_ormdantic.runtime_capabilities())
dataclass
¶
NativeCursor(description)
Cursor-like metadata returned with native query results.
NativeResult(columns, rows)
Result wrapper compatible with the serializer's cursor/row contract.
Create a result from column names and native rows.
Source code in ormdantic/engine.py
def __init__(self, columns: list[str], rows: list[tuple[Any, ...]]) -> None:
"""Create a result from column names and native rows."""
self.cursor = NativeCursor([(column,) for column in columns])
self._rows = rows
scalar()
Return the first column from the first row, if present.
Source code in ormdantic/engine.py
def scalar(self) -> Any:
"""Return the first column from the first row, if present."""
if not self._rows or not self._rows[0]:
return None
return self._rows[0][0]
NativeEngine(url)
Async facade over a persistent Rust PyNativeConnection.
Open a native connection for a database URL.
Source code in ormdantic/engine.py
def __init__(self, url: str) -> None:
"""Open a native connection for a database URL."""
if _ormdantic is None or not hasattr(_ormdantic, "PyNativeConnection"):
raise RuntimeError(
"Ormdantic vNext requires the Rust extension for native execution. "
"Install the package with maturin or reinstall the wheel."
)
self.url = url
self._connection = _ormdantic.PyNativeConnection(url)
async
¶
execute(sql, values)
Execute SQL with ordered bind values on the native connection.
Source code in ormdantic/engine.py
async def execute(self, sql: str, values: tuple[Any, ...]) -> NativeResult:
"""Execute SQL with ordered bind values on the native connection."""
loop = asyncio.get_running_loop()
result = await loop.run_in_executor(None, self._execute_sync, sql, list(values))
return NativeResult(
columns=list(result["columns"]),
rows=[tuple(row) for row in result["rows"]],
)
transaction()
Create an async transaction context manager.
Source code in ormdantic/engine.py
def transaction(self) -> NativeTransaction:
"""Create an async transaction context manager."""
return NativeTransaction(self)
async
¶
begin()
Begin a transaction on the native connection.
Source code in ormdantic/engine.py
async def begin(self) -> None:
"""Begin a transaction on the native connection."""
loop = asyncio.get_running_loop()
await loop.run_in_executor(None, self._connection.begin)
async
¶
commit()
Commit the active transaction.
Source code in ormdantic/engine.py
async def commit(self) -> None:
"""Commit the active transaction."""
loop = asyncio.get_running_loop()
await loop.run_in_executor(None, self._connection.commit)
async
¶
rollback()
Roll back the active transaction.
Source code in ormdantic/engine.py
async def rollback(self) -> None:
"""Roll back the active transaction."""
loop = asyncio.get_running_loop()
await loop.run_in_executor(None, self._connection.rollback)
NativeTransaction(engine)
Async context manager that commits or rolls back a native transaction.
Create a transaction bound to a native engine.
Source code in ormdantic/engine.py
def __init__(self, engine: NativeEngine) -> None:
"""Create a transaction bound to a native engine."""
self._engine = engine