SQLite Manager¶

SQLite database manager.

class plexosdb.db_manager.SQLiteConfig(cache_size_mb=20, mmap_size_gb=30, synchronous='NORMAL', journal_mode='WAL', foreign_keys=True, temp_store='MEMORY')¶

Bases: object

SQLite database configuration.

Parameters:

cache_size_mb (int)
mmap_size_gb (int)
synchronous (str)
journal_mode (str)
foreign_keys (bool)
temp_store (str)

cache_size_mb: int¶

mmap_size_gb: int¶

synchronous: str¶

journal_mode: str¶

foreign_keys: bool¶

temp_store: str¶

classmethod for_in_memory()¶

Create optimized config for in-memory databases.

Return type:: SQLiteConfig

classmethod for_file_database()¶

Create optimized config for file-based databases.

Return type:: SQLiteConfig

class plexosdb.db_manager.SQLiteManager(fpath_or_conn=None, *, config=None, initialize=True)¶

Bases: Generic[T]

SQLite database manager with optimized transaction support.

Parameters:

fpath_or_conn (str | Path | Connection | None)
config (SQLiteConfig | None)
initialize (bool)

property connection: Connection¶: SQLite connection.

property config: SQLiteConfig¶: SQLite configuration.

property sqlite_version: int¶: SQLite version.

property tables: list[str]¶: List of table names.

add_collation(name, callable_func)¶

Parameters:

name (str) – Name of the collation
callable_func (callable) – Function implementing the collation

Returns:

True if creation succeeded, False if it failed

Return type:

bool

backup(target_path)¶

Backup the database to a file.

Parameters:: target_path (str or Path) – Path to save the database backup
Returns:: True if backup succeeded, False if it failed
Return type:: bool

close()¶

Close the database connection and release resources.

Return type:: None

execute(query, params=None)¶

Execute a SQL statement that doesn’t return results.

Each execution is its own transaction unless used within a transaction context.

Parameters:

query (str) – SQL statement to execute
params (tuple or dict, optional) – Parameters to bind to the statement

Returns:

True if execution succeeded, False if it failed

Return type:

bool

Raises:

sqlite3.Error – If a database error occurs within a transaction

executemany(query, params_seq)¶

Execute a SQL statement with multiple parameter sets.

Parameters:

query (str) – SQL statement to execute
params_seq (list of tuples or dicts) – Sequence of parameter sets to bind

Returns:

True if execution succeeded, False if it failed

Return type:

bool

Raises:

sqlite3.Error – If a database error occurs within a transaction

executescript(script)¶

Execute a SQL script containing multiple statements.

Parameters:: script (str) – SQL script to execute
Returns:: True if execution succeeded, False if it failed
Return type:: bool

iter_query(query, params=None, batch_size=1000)¶

Execute a read-only query and return an iterator of results.

This is memory-efficient for large result sets. Use only for SELECT statements.

Parameters:

query (str) – SQL query to execute (SELECT statements only)
params (tuple or dict, optional) – Parameters to bind to the query
batch_size (int, default=1000) – Number of records to fetch in each batch

Yields:

tuple – One database row at a time

Raises:

sqlite3.Error – If a database error occurs

Return type:

Iterator[tuple[Any, …]]

last_insert_rowid()¶

Get the ID of the last inserted row.

Returns:: ID of the last inserted row
Return type:: int
Raises:: sqlite3.Error – If a database error occurs

list_table_names()¶

Return a list of current table names on the database.

Return type:: list[str]

optimize()¶

Run optimization routines on the database.

VACUUM can’t run inside a transaction, so this method may commit any pending changes before optimizing the database.

Returns:: True if optimization succeeded, False if it failed
Return type:: bool

query(query: str, params: None = None) → list[tuple[Any, ...]]¶

query(query: str, params: tuple[Any, ...] | dict[str, Any]) → list[tuple[Any, ...]]

Execute a read-only query and return all results.

Note: This method should ONLY be used for SELECT statements. For INSERT/UPDATE/DELETE, use execute() instead.

Parameters:

query (str) – SQL query to execute (SELECT statements only)
params (tuple or dict, optional) – Parameters to bind to the query

Returns:

Query results (tuples or named tuples based on initialization)

Return type:

list

Raises:

sqlite3.Error – If a database error occurs

fetchall(query, params=None)¶

Execute a query and return all results as a list of rows.

This method is a standard DB-API style alias for query().

Parameters:

query (str) – SQL query to execute (SELECT statements only)
params (tuple or dict, optional) – Parameters to bind to the query

Returns:

All rows (as tuples or named tuples based on row_factory setting)

Return type:

list

Raises:

sqlite3.Error – If a database error occurs

See also

query: Equivalent method with PlexosDB-specific naming
fetchone: Get only the first row of results
fetchall_dict: Return results as dictionaries

Examples

>>> db = SQLiteManager()
>>> db.execute("CREATE TABLE test (id INTEGER, name TEXT)")
>>> db.execute("INSERT INTO test VALUES (1, 'Alice')")
>>> db.fetchall("SELECT * FROM test")
[(1, 'Alice')]

fetchall_dict(query, params=None)¶

Execute a query and return all results as a list of dictionaries.

Parameters:

query (str) – SQL query to execute (SELECT statements only)
params (tuple or dict, optional) – Parameters to bind to the query

Returns:

All rows as dictionaries with column names as keys

Return type:

list[dict[str, Any]]

Raises:

sqlite3.Error – If a database error occurs

See also

query: Return results as tuples
iter_dicts: Memory-efficient iterator over dictionaries

Examples

>>> db = SQLiteManager()
>>> db.execute("CREATE TABLE users (id INTEGER, name TEXT)")
>>> db.execute("INSERT INTO users VALUES (1, 'Alice')")
>>> db.fetchall_dict("SELECT * FROM users")
[{'id': 1, 'name': 'Alice'}]
>>> user = db.fetchall_dict("SELECT * FROM users")[0]
>>> print(f"User {user['id']}: {user['name']}")
User 1: Alice

fetchmany(query, size=1000, params=None)¶

Execute a query and return a specified number of rows.

Parameters:

query (str) – SQL query to execute (SELECT statements only)
size (int, default=1000) – Maximum number of rows to return
params (tuple or dict, optional) – Parameters to bind to the query

Returns:

Up to ‘size’ rows from the query result

Return type:

list

Raises:

sqlite3.Error – If a database error occurs

See also

fetchall: Get all rows
fetchone: Get only one row
iter_query: Iterator over results

Examples

>>> db = SQLiteManager()
>>> db.execute("CREATE TABLE items (id INTEGER, name TEXT)")
>>> db.executemany("INSERT INTO items VALUES (?, ?)", [(i, f"Item {i}") for i in range(1, 101)])
>>> # Get first 10 items
>>> first_batch = db.fetchmany("SELECT * FROM items", size=10)
>>> len(first_batch)
10

fetchone(query, params=None)¶

Execute a query and return only the first result row.

Parameters:

query (str) – SQL query to execute (SELECT statements only)
params (tuple or dict, optional) – Parameters to bind to the query

Returns:

First row of results or None if no results

Return type:

tuple or namedtuple or None

Raises:

sqlite3.Error – If a database error occurs

See also

fetchall: Get all rows
fetchone_dict: Get first row as dictionary

Examples

>>> db = SQLiteManager()
>>> db.execute("CREATE TABLE users (id INTEGER, name TEXT)")
>>> db.execute("INSERT INTO users VALUES (1, 'Alice')")
>>> user = db.fetchone("SELECT * FROM users WHERE id = ?", (1,))
>>> user
(1, 'Alice')
>>> user = db.fetchone("SELECT * FROM users WHERE id = ?", (999,))
>>> user is None
True

fetchone_dict(query, params=None)¶

Execute a query and return only the first result row as a dictionary.

Parameters:

query (str) – SQL query to execute (SELECT statements only)
params (tuple or dict, optional) – Parameters to bind to the query

Returns:

First row as dictionary with column names as keys, or None if no results

Return type:

dict[str, Any] or None

Raises:

sqlite3.Error – If a database error occurs

See also

fetchone: Get first row as tuple
fetchall_dict: Get all rows as dictionaries

Examples

>>> db = SQLiteManager()
>>> db.execute("CREATE TABLE users (id INTEGER, name TEXT, email TEXT)")
>>> db.execute("INSERT INTO users VALUES (1, 'Alice', 'alice@example.com')")
>>> user = db.fetchone_dict("SELECT * FROM users WHERE id = ?", (1,))
>>> print(f"{user['name']}'s email is {user['email']}")
Alice's email is alice@example.com

iter_dicts(query, params=None, batch_size=1000)¶

Execute a read-only query and yield results as dictionaries.

This is memory-efficient for large result sets. Each row is returned as a dictionary with column names as keys.

Parameters:

query (str) – SQL query to execute (SELECT statements only)
params (tuple or dict, optional) – Parameters to bind to the query
batch_size (int, default=1000) – Number of records to fetch in each batch

Yields:

dict[str, Any] – One database row at a time as a dictionary

Raises:

sqlite3.Error – If a database error occurs

Return type:

Iterator[dict[str, Any]]

See also

iter_query: Iterator over tuples or named tuples
fetchall_dict: Get all results as dictionaries at once

Examples

>>> db = SQLiteManager()
>>> db.execute("CREATE TABLE users (id INTEGER, name TEXT)")
>>> db.executemany("INSERT INTO users VALUES (?, ?)", [(i, f"User {i}") for i in range(1, 1001)])
>>> # Process users efficiently one at a time
>>> for user in db.iter_dicts("SELECT * FROM users"):
...     print(f"Processing user {user['name']}")

transaction()¶

Begin a transaction that can span multiple operations.

This provides explicit transaction control for grouping multiple operations into a single atomic unit that either all succeed or all fail.

Returns:: A context manager for the transaction
Return type:: context manager

Examples

>>> with db.transaction():
>>>     db.execute("INSERT INTO table1 VALUES (1, 'a')")
>>>     db.execute("INSERT INTO table2 VALUES (1, 'b')")

Raises:: sqlite3.Error – If a database error occurs during transaction

insert_records(table_name, records)¶

Insert records into a table using dictionaries with column names as keys.

Fails if duplicate records exist (uses INSERT without OR REPLACE).

Parameters:

table_name (str) – Name of the table to insert records into
records (dict or list of dicts) – Dictionary or list of dictionaries with column names as keys

Returns:

True if insertion succeeded, False if it failed

Return type:

bool

Raises:

ValueError – If table doesn’t exist, records are empty, or have inconsistent keys
sqlite3.Error – If a database error occurs within a transaction or duplicate records exist