`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.
`TableNames`	Base class for app-scoped SQLite table name collections.

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.
`sanitize_table_prefix`	Sanitize an app_id for safe use as a SQLite table name prefix.
`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.sanitize_table_prefix(app_id: str) → str[source]¶

Sanitize an app_id for safe use as a SQLite table name prefix.

Replaces any character that is not alphanumeric or underscore with an underscore, prepends an underscore if the result starts with a digit, and always appends an 8-character hash of the original app_id.

The hash prevents collisions when two different app_ids sanitize to the same string (e.g. my-app and my_app), and also protects against a user accidentally choosing an app_id that looks like a previously sanitized value.

Parameters:: app_id (str) – The application identifier to sanitize
Returns:: A string safe for use in SQLite table names

class pynenc.util.sqlite_utils.TableNames(app_id: str, component: str)[source]¶

Base class for app-scoped SQLite table name collections.

Subclasses call super().__init__(app_id, component) where component is a short label such as "broker" or "state_backend". The resulting

Attr:: table_prefix is the single source of truth used both when creating tables and when purging them, eliminating duplicated prefix strings.

Initialization

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.