TableSync Module

This module handles table synchronization between Pydantic models and SQLite.

class wsqlite.core.sync.TableSync(model, db_path: str, table_name: str | None = None)

Bases: object

Handles table synchronization between Pydantic models and SQLite (sync).

__init__(model, db_path: str, table_name: str | None = None)

Initialize table sync.

Parameters:

• model – Pydantic BaseModel class.

• db_path – Path to SQLite database file.

• table_name – Optional custom table name.

create_if_not_exists()

Create the table if it doesn’t exist, handling FTS5 virtual tables.

sync_with_model()

Sync the table with the Pydantic model, adding new columns if necessary.

table_exists() → bool

Check if the table exists in the database.

drop_table()

Drop the table from the database.

get_columns() → list[str]

Get list of column names in the table.

create_index(columns: list[str], index_name: str | None = None, unique: bool = False)

Create an index on the specified columns.

drop_index(index_name: str)

Drop an index from the table.

get_indexes() → list[dict]

Get list of indexes on the table.

AsyncTableSync

class wsqlite.core.sync.AsyncTableSync(model, db_path: str, table_name: str | None = None)

Bases: object

Handles table synchronization between Pydantic models and SQLite (async).

__init__(model, db_path: str, table_name: str | None = None)

Initialize async table sync.

async create_if_not_exists_async()

Create the table if it doesn’t exist (async).

async sync_with_model_async()

Sync the table with the Pydantic model, adding new columns if necessary (async).

async table_exists_async() → bool

Check if the table exists in the database (async).

async drop_table_async()

Drop the table from the database (async).

async get_columns_async() → list[str]

Get list of column names in the table (async).

async create_index_async(columns: list[str], index_name: str | None = None, unique: bool = False)

Create an index on the specified columns (async).

async drop_index_async(index_name: str)

Drop an index from the table (async).

async get_indexes_async() → list[dict]

Get list of indexes on the table (async).

Methods Overview

Method	Description
create_if_not_exists()	Create table if it doesn’t exist
sync_with_model()	Sync table with model (add new columns)
table_exists()	Check if table exists
drop_table()	Drop the table
get_columns()	Get list of column names
create_index()	Create an index
drop_index()	Drop an index
get_indexes()	Get list of indexes

Example Usage

from pydantic import BaseModel
from wsqlite import TableSync

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

sync = TableSync(User, "database.db")

# Create table
sync.create_if_not_exists()

# Check if exists
if sync.table_exists():
    print("Table exists!")

# Get columns
columns = sync.get_columns()
print(f"Columns: {columns}")

# Create index
sync.create_index(["email"], unique=True)

# Get indexes
indexes = sync.get_indexes()
print(f"Indexes: {indexes}")