SqliterDB

The main entry point for all database operations. Manages SQLite connections, table creation, CRUD operations, caching, and transactions.

from sqliter import SqliterDB

Source: sqliter/sqliter.py

See also: Guide -- Connecting, Guide -- Properties, Guide -- Tables, Guide -- Data Operations, Guide -- Caching, Guide -- Transactions

Class Attribute

MEMORY_DB

Constant for in-memory database filename.

MEMORY_DB = ":memory:"

Constructor

def __init__(
    self,
    db_filename: str | None = None,
    *,
    memory: bool = False,
    auto_commit: bool = True,
    debug: bool = False,
    logger: logging.Logger | None = None,
    reset: bool = False,
    return_local_time: bool = True,
    cache_enabled: bool = False,
    cache_max_size: int = 1000,
    cache_ttl: int | None = None,
    cache_max_memory_mb: int | None = None,
) -> None:

Parameters:

Raises:

• ValueError -- If no filename is provided for a non-memory database, if cache_max_size <= 0, if cache_ttl < 0, or if cache_max_memory_mb <= 0.

Example:

# File-based database
db = SqliterDB("myapp.db")

# In-memory database
db = SqliterDB(memory=True)

# With caching enabled
db = SqliterDB(
    "myapp.db",
    cache_enabled=True,
    cache_ttl=300,
    cache_max_memory_mb=50,
)

Properties

filename

Returns the database filename, or None if using an in-memory database.

is_memory

Returns True if the database is in-memory.

is_autocommit

Returns True if auto-commit is enabled.

is_connected

Returns True if a database connection is currently open.

table_names

Returns a list of all table names in the database. Temporarily connects if not already connected and restores the connection state afterward.

Example:

db = SqliterDB("myapp.db")
print(db.table_names)  # ["users", "posts", ...]

Connection Methods

connect()

Establish a connection to the SQLite database. Enables foreign key constraint enforcement via PRAGMA foreign_keys = ON.

def connect(self) -> sqlite3.Connection:

Returns:

sqlite3.Connection -- The SQLite connection object.

Raises:

• DatabaseConnectionError -- If unable to connect.

close()

Close the database connection. Commits pending changes if auto_commit is True. Clears the query cache and resets cache statistics.

def close(self) -> None:

commit()

Explicitly commit the current transaction.

def commit(self) -> None:

Table Methods

create_table()

Create a database table based on a model class. Handles column definitions, primary keys, foreign key constraints, and indexes.

def create_table(
    self,
    model_class: type[BaseDBModel],
    *,
    exists_ok: bool = True,
    force: bool = False,
) -> None:

Parameters:

Raises:

• TableCreationError -- If there is an error creating the table.
• InvalidIndexError -- If index fields do not exist in the model.

Example:

db.create_table(User)
db.create_table(User, force=True)  # Drop and recreate

drop_table()

Drop the table associated with a model class.

def drop_table(
    self,
    model_class: type[BaseDBModel],
) -> None:

Parameters:

Raises:

• TableDeletionError -- If there is an error dropping the table.

CRUD Methods

insert()

Insert a new record into the database. Sets created_at and updated_at timestamps automatically.

def insert(
    self,
    model_instance: T,
    *,
    timestamp_override: bool = False,
) -> T:

Parameters:

Returns:

T -- A new model instance with pk set to the inserted row's ID.

Raises:

• RecordInsertionError -- If there is an error during insertion.
• ForeignKeyConstraintError -- If a FK value does not exist in the referenced table.

Example:

user = User(name="Alice", email="[email protected]")
saved = db.insert(user)
print(saved.pk)  # e.g. 1

bulk_insert()

Insert multiple records of the same model type in a single transaction. More efficient than calling insert() in a loop.

def bulk_insert(
    self,
    instances: Sequence[T],
    *,
    timestamp_override: bool = False,
) -> list[T]:

Parameters:

Returns:

list[T] -- New model instances with pk set for each inserted row. Returns [] for an empty input sequence.

Raises:

• RecordInsertionError -- If there is an error during insertion.
• ForeignKeyConstraintError -- If a FK value does not exist in the referenced table.
• ValueError -- If instances contain mixed model types.

Behavior:

• All records are inserted within a single transaction. If any record fails, the entire batch is rolled back.
• When called inside a with db: context manager, the commit is deferred to context exit.
• Cache is invalidated once after all records are inserted.

Example:

users = [
    User(name="Alice", email="[email protected]"),
    User(name="Bob", email="[email protected]"),
]
saved = db.bulk_insert(users)
print(saved[0].pk, saved[1].pk)  # e.g. 1, 2

get()

Retrieve a single record by its primary key.

def get(
    self,
    model_class: type[T],
    primary_key_value: int,
    *,
    bypass_cache: bool = False,
    cache_ttl: int | None = None,
) -> T | None:

Parameters:

Returns:

T | None -- The model instance if found, None otherwise.

Raises:

• RecordFetchError -- If there is an error fetching the record.
• ValueError -- If cache_ttl is negative.

Example:

user = db.get(User, 1)
if user:
    print(user.name)

Caching:

When cache_enabled=True, get() uses the query cache. Use bypass_cache=True to force a fresh lookup, or cache_ttl to override the global cache TTL for this call.

update()

Update an existing record. The model instance must have a valid pk. Automatically sets updated_at to the current time.

def update(
    self,
    model_instance: BaseDBModel,
) -> None:

Parameters:

Raises:

• RecordUpdateError -- If there is an error updating the record.
• RecordNotFoundError -- If no record matches the pk.

Example:

user = db.get(User, 1)
user.name = "Bob"
db.update(user)

delete()

Delete a record by its primary key.

def delete(
    self,
    model_class: type[BaseDBModel],
    primary_key_value: int | str,
) -> None:

Parameters:

Raises:

• RecordDeletionError -- If there is an error deleting the record.
• RecordNotFoundError -- If no record matches the pk.
• ForeignKeyConstraintError -- If the record is still referenced (with RESTRICT).

Example:

db.delete(User, 1)

select()

Create a QueryBuilder for constructing queries with filters, ordering, pagination, and more.

def select(
    self,
    model_class: type[T],
    fields: list[str] | None = None,
    exclude: list[str] | None = None,
) -> QueryBuilder[T]:

Parameters:

Returns:

QueryBuilder[T] -- A query builder for method chaining.

Example:

# Simple query
users = db.select(User).fetch_all()

# With filters and ordering
users = (
    db.select(User)
    .filter(active=True)
    .order("name")
    .limit(10)
    .fetch_all()
)

# Exclude fields
users = db.select(User, exclude=["password_hash"]).fetch_all()

Cache Methods

get_cache_stats()

Get cache performance statistics.

def get_cache_stats(self) -> dict[str, int | float]:

Returns:

dict[str, int | float] -- Dictionary with keys:

Example:

stats = db.get_cache_stats()
print(f"Hit rate: {stats['hit_rate']}%")

clear_cache()

Clear all cached query results. Cache statistics (hits/misses) are preserved.

def clear_cache(self) -> None:

Context Manager

SqliterDB can be used as a context manager for transaction management. Within a with block, auto-commit is suppressed and all operations are wrapped in a single transaction.

def __enter__(self) -> Self:
def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc_value: BaseException | None,
    traceback: TracebackType | None,
) -> None:

Behavior:

• __enter__: Opens a connection and begins a transaction.
• __exit__ (no exception): Commits the transaction and closes the connection.
• __exit__ (exception raised): Rolls back the transaction and closes the connection.
• Cache is cleared on exit in both cases.

Example:

db = SqliterDB("myapp.db")

with db:
    db.create_table(User)
    db.insert(User(name="Alice"))
    db.insert(User(name="Bob"))
    # Both inserts are committed together on exit

# If an exception occurs, both inserts are rolled back

See also: Guide -- Transactions