tessl/pypi-aiosqlite

asyncio bridge to the standard sqlite3 module

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Connection Management

Name: tessl/pypi-aiosqlite
Author: tessl

Database connection functionality including connection creation, lifecycle management, and resource cleanup. Provides async context managers for automatic resource management and configuration of connection properties.

Capabilities

Connection Creation

Creates an async database connection using a connector function that runs in a separate thread.

def connect(
    database: Union[str, Path],
    *,
    iter_chunk_size: int = 64,
    loop: Optional[asyncio.AbstractEventLoop] = None,
    **kwargs: Any
) -> Connection:
    """
    Create and return a connection proxy to the sqlite database.
    
    Parameters:
    - database: Path to the database file (str or pathlib.Path)
    - iter_chunk_size: Number of rows to fetch at once during iteration (default: 64)
    - loop: Deprecated, ignored for compatibility
    - **kwargs: Additional arguments passed to sqlite3.connect()
    
    Returns:
    Connection: Async database connection object
    """

Usage example:

import aiosqlite

# Basic connection
conn = aiosqlite.connect("database.db")

# With connection parameters
conn = aiosqlite.connect(
    "database.db",
    iter_chunk_size=100,
    timeout=10.0,
    check_same_thread=False
)

# Using context manager (recommended)
async with aiosqlite.connect("database.db") as db:
    # Database operations here
    pass

Connection Lifecycle

The Connection class manages the database connection lifecycle and thread communication.

class Connection:
    """
    Async SQLite database connection running in separate thread.
    """
    
    def __init__(
        self,
        connector: Callable[[], sqlite3.Connection],
        iter_chunk_size: int,
        loop: Optional[asyncio.AbstractEventLoop] = None
    ) -> None: ...
    
    async def __aenter__(self) -> "Connection":
        """Async context manager entry."""
        
    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
        """Async context manager exit with automatic cleanup."""
    
    def __await__(self) -> Generator[Any, None, "Connection"]:
        """Enable awaiting the connection directly."""

Connection Control

Methods for controlling connection state and lifecycle.

async def close(self) -> None:
    """
    Complete queued queries/cursors and close the connection.
    
    Ensures all pending operations complete before closing.
    After calling close(), the connection cannot be used for new operations.
    """

async def interrupt(self) -> None:
    """
    Interrupt pending queries.
    
    Calls sqlite3.Connection.interrupt() to cancel long-running operations.
    Operations may still complete normally if interruption occurs too late.
    """

Usage example:

# Manual connection management
conn = await aiosqlite.connect("database.db")
try:
    # Database operations
    await conn.execute("SELECT * FROM large_table")
    # Interrupt if needed
    await conn.interrupt()
finally:
    await conn.close()

Connection Properties

Properties for accessing connection state and configuration.

@property
def in_transaction(self) -> bool:
    """
    True if connection has an uncommitted transaction.
    
    Returns:
    bool: Transaction state
    """

@property
def isolation_level(self) -> Optional[str]:
    """
    Current isolation level setting.
    
    Returns:
    Optional[str]: 'DEFERRED', 'IMMEDIATE', 'EXCLUSIVE', or None for autocommit
    """

@isolation_level.setter
def isolation_level(self, value: IsolationLevel) -> None:
    """
    Set isolation level for transactions.
    
    Parameters:
    - value: 'DEFERRED', 'IMMEDIATE', 'EXCLUSIVE', or None for autocommit
    """

@property
def row_factory(self) -> Optional[type]:
    """
    Current row factory for query results.
    
    Returns:
    Optional[type]: Row factory class or None for default tuple rows
    """

@row_factory.setter
def row_factory(self, factory: Optional[type]) -> None:
    """
    Set row factory for query results.
    
    Parameters:
    - factory: Row factory class (e.g., sqlite3.Row) or None for tuples
    """

@property
def text_factory(self) -> Callable[[bytes], Any]:
    """
    Current text factory for handling TEXT columns.
    
    Returns:
    Callable: Function to convert bytes to text
    """

@text_factory.setter
def text_factory(self, factory: Callable[[bytes], Any]) -> None:
    """
    Set text factory for TEXT column handling.
    
    Parameters:
    - factory: Function to convert bytes to desired text type
    """

@property
def total_changes(self) -> int:
    """
    Total number of database changes since connection opened.
    
    Returns:
    int: Total change count
    """

Usage example:

async with aiosqlite.connect("database.db") as db:
    # Configure connection
    db.row_factory = aiosqlite.Row  # Get rows as dict-like objects
    db.isolation_level = "IMMEDIATE"  # Set transaction isolation
    
    # Check connection state
    print(f"In transaction: {db.in_transaction}")
    print(f"Total changes: {db.total_changes}")

Install with Tessl CLI