`pynenc.util.sqlite_utils`¶

SQLite utilities for Pynenc shared state management.

Provides small helpers for creating connections and simple cross-process operations used by the test orchestrator/state components.

Module Contents¶

Classes¶

SQLiteConnection

A wrapper for sqlite3.Connection that adds retry logic to execute method.

Functions¶

`create_sqlite_connection`	Create and return a configured sqlite3.Connection for concurrent test use.
`get_sqlite_sqlite_db_path`	Get and validate the SQLite database path.
`delete_tables_with_prefix`	Delete all data from tables in the SQLite database that start with the given prefix.

Data¶

logger

API¶

pynenc.util.sqlite_utils.logger¶: ‘getLogger(…)’

class pynenc.util.sqlite_utils.SQLiteConnection(conn: sqlite3.Connection)[source]¶

A wrapper for sqlite3.Connection that adds retry logic to execute method.

This wrapper delegates all methods to the underlying connection, but overrides execute to include exponential backoff retry on database lock errors.

Initialization

execute(sql: str, parameters: tuple[Any, ...] = (), /) → sqlite3.Cursor[source]¶

Execute a SQL query with exponential backoff retry on database lock errors.

Parameters:

sql – SQL query string
parameters – Query parameters

Returns:

Cursor from the successful execution

__getattr__(name: str) → Any[source]¶: Delegate all other attributes to the underlying connection.

__enter__() → pynenc.util.sqlite_utils.SQLiteConnection[source]¶: Enter context manager.

__exit__(exc_type: type | None, exc_val: Any, exc_tb: Any) → None[source]¶: Exit context manager.

pynenc.util.sqlite_utils.create_sqlite_connection(sqlite_db_path: str | pathlib.Path) → pynenc.util.sqlite_utils.SQLiteConnection[source]¶

Create and return a configured sqlite3.Connection for concurrent test use.

The connection uses WAL journal mode and a busy timeout so concurrent clients are less likely to raise transient “database is locked” errors.

Parameters:: sqlite_db_path – Path to the SQLite database file
Returns:: A configured sqlite3.Connection

pynenc.util.sqlite_utils.get_sqlite_sqlite_db_path(sqlite_db_path: str) → str[source]¶

Get and validate the SQLite database path.

Parameters:: sqlite_db_path – The configured database path
Returns:: The validated database path
Raises:: ValueError – If no database path is configured

pynenc.util.sqlite_utils.delete_tables_with_prefix(sqlite_db_path: str | pathlib.Path, prefix: str) → None[source]¶

Delete all data from tables in the SQLite database that start with the given prefix.

Uses a short-lived connection and closes cursors promptly to avoid holding read cursors open which can cause ‘database is locked’ when other clients write.