API Reference

Base

Backend-agnostic base for stored-procedure / SQL executors.

Template Method pattern: this class owns everything that’s the same across DB backends (cursor lifecycle, commit/rollback, fetch-to-dict, chunked streaming, error wrapping). A subclass only has to implement:

• _connect() -> open a DB-API 2.0 connection

• _call_procedure_sql() -> dialect-specific “call this SP” string

Everything else (call_procedure, execute, stream, stream_procedure) is inherited for free. See sqlserver.py for a ~15-line concrete example.

exception sql_executor.base.DatabaseError

Bases: Exception

Raised when a stored procedure / SQL execution fails.

Callers should catch this rather than checking for None/False — it’s always raised on failure, never swallowed.

class sql_executor.base.BaseSqlExecutor

Bases: ABC

driver_error

alias of Exception

call_procedure(sp_name, params=(), fetch=False)

Parameters:

• sp_name (str)

• params (Sequence[Any])

• fetch (bool)

execute(sql, params=(), fetch=False)

Run raw SQL (SELECT/INSERT/UPDATE/DELETE).

Parameters:

• sql (str)

• params (Sequence[Any])

• fetch (bool)

stream(sql, params=(), chunk_size=500)

Yield rows in chunks — for big SELECTs or SPs returning lots of rows.

Parameters:

• sql (str)

• params (Sequence[Any])

• chunk_size (int)

Return type:

Iterator[Dict[str, Any]]

stream_procedure(sp_name, params=(), chunk_size=500)

Parameters:

• sp_name (str)

• params (Sequence[Any])

• chunk_size (int)

Return type:

Iterator[Dict[str, Any]]

SQL Server backend

SQL Server backend — the only piece that knows about pyodbc.

class sql_executor.sqlserver.SqlServerExecutor(connstring)

Bases: BaseSqlExecutor

Stored-procedure / SQL executor for SQL Server, backed by pyodbc.

Parameters:

connstring (str)

driver_error

alias of Error

Fake executor (testing)

In-memory test double — no real DB connection involved.

class sql_executor.fake.FakeExecutor

Bases: BaseSqlExecutor

Register a Python callable per stored-procedure name and call it just like the real thing, for unit tests.

fake = FakeExecutor() fake.register(“dbo.GetIncidents”, lambda since: [{“id”: 1}]) fake.call_procedure(“dbo.GetIncidents”, params=(“2026-06-01”,), fetch=True)

driver_error

alias of Exception

register(sp_name, handler)

Register a virtual subclass of an ABC.

Returns the subclass, to allow usage as a class decorator.

Parameters:

• sp_name (str)

• handler (Callable[[...], Any])

call_procedure(sp_name, params=(), fetch=False)

Parameters:

• sp_name (str)

• params (Sequence[Any])

• fetch (bool)

execute(sql, params=(), fetch=False)

Run raw SQL (SELECT/INSERT/UPDATE/DELETE).

Parameters:

• sql (str)

• params (Sequence[Any])

• fetch (bool)