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)

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)

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.

• min_pool_size – Minimum number of connections in pool.

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

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.

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

Get records with pagination.

Parameters:

• limit – Maximum number of records to return.

• offset – Number of records to skip.

• order_by – Column to order by.

• order_desc – If True, order descending.

Returns:

List of model instances.

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.

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).

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).

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)