Connection Module

This module provides connection management for SQLite databases.

Transaction

class wsqlite.core.connection.Transaction(db_path: str)

Bases: object

Context manager for database transactions (sync).

__init__(db_path: str)

commit()

rollback()

execute(query: str, values: tuple = None) → Any

AsyncTransaction

class wsqlite.core.connection.AsyncTransaction(db_path: str)

Bases: object

Context manager for database transactions (async).

__init__(db_path: str)

async commit()

async rollback()

async execute(query: str, values: tuple = None) → Any

Connection Functions

wsqlite.core.connection.get_connection(db_path: str) → _SQLiteConnection

Get a connection from global pool (sync).

Usage:

with get_connection(“database.db”) as conn:

conn.execute(…)

async wsqlite.core.connection.get_async_connection(db_path: str) → Any

Get a configured async connection.

Usage:

async with get_async_connection(“database.db”) as conn:

await conn.execute(…)

wsqlite.core.connection.get_transaction(db_path: str) → Iterator[Transaction]

Get a transaction context manager (sync).

wsqlite.core.connection.get_async_transaction(db_path: str) → AsyncTransaction

Get an async transaction context manager.

wsqlite.core.connection.close_global_connection()

Close global connection.

Exception Handling

Connection-related exceptions are defined in wsqlite.exceptions:

• wsqlite.exceptions.ConnectionError

• wsqlite.exceptions.TransactionError

Example Usage

from wsqlite.core.connection import get_connection, Transaction

# Using connection context manager
with get_connection("database.db") as conn:
    cursor = conn.execute("SELECT COUNT(*) FROM users")
    count = cursor.fetchone()[0]
    print(f"Total users: {count}")

# Using transaction
with Transaction("database.db") as txn:
    txn.execute("INSERT INTO users (name) VALUES (?)", ("Alice",))
    txn.commit()