WSQLite Repository

The main repository class for SQLite database operations.

class wsqlite.core.repository.WSQLite(model: type[BaseModel], db_path: str, pool_size: int = 10, min_pool_size: int = 2, use_pool: bool = True, table_name: str | None = None, soft_delete: bool = False, deleted_at_field: str = 'deleted_at', pool: ConnectionPool | None = None, sync_handler: TableSync | None = None)

Bases: object

SQLite repository using Pydantic models.

Provides a simple interface for CRUD operations on SQLite tables, with automatic table creation, schema synchronization, and connection pooling.

Example

from pydantic import BaseModel from wsqlite import WSQLite

class User(BaseModel):

id: int name: str email: str

db = WSQLite(User, “database.db”) db.insert(User(id=1, name=”John”, email=”john@example.com”))

__init__(model: type[BaseModel], db_path: str, pool_size: int = 10, min_pool_size: int = 2, use_pool: bool = True, table_name: str | None = None, soft_delete: bool = False, deleted_at_field: str = 'deleted_at', pool: ConnectionPool | None = None, sync_handler: TableSync | None = None)

Initialize the repository with a Pydantic model.

Parameters:

• model – Pydantic BaseModel class defining the table schema.

• db_path – Path to SQLite database file.

• pool_size – Maximum number of connections in pool (if pool is not provided).

• min_pool_size – Minimum number of connections in pool (if pool is not provided).

• use_pool – Whether to use connection pooling (recommended).

• table_name – Optional custom table name.

• soft_delete – Whether to use soft deletes (default False).

• deleted_at_field – Name of the field for soft deletes (default “deleted_at”).

• pool – Optional pre-configured connection pool.

• sync_handler – Optional pre-configured TableSync instance.

insert(data: BaseModel) → None

Insert a new record into the database.

get_all() → list[BaseModel]

Get all records from the table.

get_by_field(**filters) → list[BaseModel]

Get records filtered by specified fields.

update(record_id: int, data: BaseModel) → None

Update a record in the database.

delete(record_id: int) → None

Delete a record from the database (hard or soft).

restore(record_id: int) → None

Restore a soft-deleted record.

load_related(instance: BaseModel, attribute_name: str, related_db: WSQLite, foreign_key: str, local_key: str = 'id', is_list: bool = True)

Load related data from another table onto a model instance.

Parameters:

• instance – The model instance to enrich.

• attribute_name – The name of the attribute to set on the instance.

• related_db – The WSQLite instance for the related table.

• foreign_key – The foreign key column name on the related table.

• local_key – The local key column name on the current instance’s table.

• is_list – True for one-to-many relationships, False for many-to-one.

async load_related_async(instance: BaseModel, attribute_name: str, related_db: WSQLite, foreign_key: str, local_key: str = 'id', is_list: bool = True)

Load related data from another table onto a model instance (async).

get_paginated(limit: int = 10, offset: int = 0, order_by: str | None = None, order_desc: bool = False) → list[BaseModel]

Get records with pagination.

get_page(page: int = 1, per_page: int = 10) → list[BaseModel]

Get records by page number.

Parameters:

• page – Page number (1-indexed).

• per_page – Number of records per page.

Returns:

List of model instances for the requested page.

count() → int

Get total number of records in the table.

insert_many(data_list: list[BaseModel]) → None

Insert multiple records in a single transaction.

Parameters:

data_list – List of model instances to insert.

update_many(updates: list[tuple[BaseModel, int]]) → int

Update multiple records.

Parameters:

updates – List of (model, record_id) tuples.

Returns:

Number of records updated.

delete_many(record_ids: list[int]) → int

Delete multiple records by their IDs (hard or soft).

Parameters:

record_ids – List of record IDs to delete.

Returns:

Number of records deleted.

execute_transaction(operations: list[tuple[str, tuple]]) → list[Any]

Execute multiple operations in a transaction.

Parameters:

operations – List of (query, params) tuples.

Returns:

List of results from each operation.

with_transaction(func: Callable[[Transaction], Any]) → Any

Execute a function within a transaction.

Parameters:

func – Function that receives Transaction and performs operations.

Returns:

Result of the function.

insert_with_retry(**kwargs)

async insert_async(data: BaseModel) → None

Insert a new record into the database (async).

async get_all_async() → list[BaseModel]

Get all records from the table (async).

async get_by_field_async(**filters) → list[BaseModel]

Get records filtered by specified fields (async).

async update_async(record_id: int, data: BaseModel) → None

Update a record in the database (async).

async delete_async(record_id: int) → None

Delete a record from the database (async, hard or soft).

async restore_async(record_id: int) → None

Restore a soft-deleted record (async).

async search_async(query: str, order_by_rank: bool = True) → list[BaseModel]

Perform a full-text search on an FTS5 table (async).

Parameters:

• query – The search query string.

• order_by_rank – Whether to sort results by relevance (default True).

Returns:

A list of matching model instances.

async get_paginated_async(limit: int = 10, offset: int = 0, order_by: str | None = None, order_desc: bool = False) → list[BaseModel]

Get records with pagination (async).

async get_page_async(page: int = 1, per_page: int = 10) → list[BaseModel]

Get records by page number (async).

async count_async() → int

Get total number of records in the table (async).

async insert_many_async(data_list: list[BaseModel]) → None

Insert multiple records in a single transaction (async).

async update_many_async(updates: list[tuple[BaseModel, int]]) → int

Update multiple records (async).

async delete_many_async(record_ids: list[int]) → int

Delete multiple records by their IDs (async, hard or soft).

async execute_transaction_async(operations: list[tuple[str, tuple]]) → list[Any]

Execute multiple operations in a transaction (async).

async with_transaction_async(func: Callable[[AsyncTransaction], Any]) → Any

Execute a function within a transaction (async).

Sync Methods

CRUD Operations

Method	Description
insert()	Insert a single record
get_all()	Retrieve all records
get_by_field()	Filter records by field values
update()	Update a record by ID
delete()	Delete a record by ID

Pagination

Method	Description
get_paginated()	Get records with limit/offset
get_page()	Get records by page number
count()	Get total record count

Bulk Operations

Method	Description
insert_many()	Insert multiple records
update_many()	Update multiple records
delete_many()	Delete multiple records

Transactions

Method	Description
execute_transaction()	Execute operations in transaction
with_transaction()	Execute callback in transaction

Async Methods

All sync methods have async equivalents with _async suffix:

• insert_async()

• get_all_async()

• get_by_field_async()

• update_async()

• delete_async()

• get_paginated_async()

• get_page_async()

• count_async()

• insert_many_async()

• update_many_async()

• delete_many_async()

• execute_transaction_async()

• with_transaction_async()

Example Usage

from pydantic import BaseModel
from wsqlite import WSQLite

class User(BaseModel):
    id: int
    name: str
    email: str

db = WSQLite(User, "database.db")

# Insert
db.insert(User(id=1, name="Alice", email="alice@example.com"))

# Query
users = db.get_all()
alice_users = db.get_by_field(name="Alice")

# Update
db.update(1, User(id=1, name="Alice Smith", email="alice@example.com"))

# Delete
db.delete(1)