semolina.pool

MockPool for testing queries without a real warehouse connection.

MockPool provides the same connection interface as adbc-poolhouse pools (pool.connect() context manager -> connection.cursor()), enabling SemolinaCursor (Phase 26) and _Query.execute() to work identically with both mock and real pools.

Classes

MockPool	In-memory pool for testing without warehouse connections.
MockConnection	In-memory connection wrapping fixture data.
MockCursor	DBAPI 2.0-compatible cursor backed by in-memory fixture data.

Module Contents

class semolina.pool.MockPool

In-memory pool for testing without warehouse connections.

Stores fixture data keyed by view name. Provides the same pool.connect() -> connection.cursor() interface that adbc-poolhouse pools expose, so downstream code works identically with mock and real pools.

load(view_name: str, data: list[dict[str, Any]]) → None

Load fixture data for a semantic view.

Parameters:

• view_name – View name matching SemanticView’s view parameter.

• data – List of row dicts with field names as keys.

connect() → MockConnection

Return a mock connection (matches adbc-poolhouse pool.connect()).

Returns:

MockConnection wrapping this pool’s fixtures.

close() → None

No-op for mock (real pools release resources here).

class semolina.pool.MockConnection(fixtures: dict[str, list[dict[str, Any]]])

In-memory connection wrapping fixture data.

cursor() → MockCursor

Return a MockCursor backed by this connection’s fixtures.

Returns:

MockCursor instance.

close() → None

No-op for mock connections.

class semolina.pool.MockCursor(fixtures: dict[str, list[dict[str, Any]]])

DBAPI 2.0-compatible cursor backed by in-memory fixture data.

Receives _Query objects via _execute_query() for predicate evaluation. Returns results in DBAPI format: fetchall() returns list[tuple], description returns 7-element tuples.

property description: list[tuple[Any, Ellipsis]] | None

list of 7-element tuples, or None before execute.

Returns:

List of (name, None, None, None, None, None, None) tuples after _execute_query, or None before any query has been executed.

Type:

DBAPI 2.0 description

property rowcount: int

number of rows from last execute.

Returns:

Number of rows in the current result set.

Type:

DBAPI 2.0 rowcount

execute(sql: str, params: Any = None) → None

DBAPI 2.0 execute – populate results from fixture data.

Extracts the view name from the SQL FROM clause and loads that view’s fixture data. Does not parse predicates; returns all rows for the matched view.

Parameters:

• sql – SQL string containing a FROM clause with the view name.

• params – Bind parameters (accepted for DBAPI compliance, unused).

fetchall() → list[tuple[Any, Ellipsis]]

Fetch all remaining rows as list of tuples.

Returns:

List of tuple rows. Empty list if no rows remain.

fetchone() → tuple[Any, Ellipsis] | None

Fetch the next row, or None if exhausted.

Returns:

Single tuple row, or None if no rows remain.

fetchmany(size: int = 1) → list[tuple[Any, Ellipsis]]

Fetch up to size rows as tuples, advancing the cursor position.

Parameters:

size – Maximum number of rows to return. Defaults to 1.

Returns:

List of tuple rows (may be shorter than size).

close() → None

No-op for mock cursors.