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

Query Operations

Name: tessl/pypi-aiosqlite
Author: tessl

Comprehensive query execution methods for SQL operations including single queries, batch operations, and data retrieval. All methods support async/await syntax and provide automatic cursor management through context managers.

Capabilities

Single Query Execution

Execute individual SQL statements with optional parameter binding for safe query execution.

async def execute(
    self, 
    sql: str, 
    parameters: Optional[Iterable[Any]] = None
) -> Cursor:
    """
    Helper to create a cursor and execute the given query.
    
    Parameters:
    - sql: SQL statement to execute
    - parameters: Optional parameter values for query placeholders
    
    Returns:
    Cursor: Async cursor with query results
    """

Usage example:

async with aiosqlite.connect("database.db") as db:
    # Simple query without parameters
    async with db.execute("CREATE TABLE users (id INTEGER, name TEXT)") as cursor:
        pass
    
    # Query with parameters (recommended for user input)
    async with db.execute(
        "INSERT INTO users (id, name) VALUES (?, ?)",
        (1, "John Doe")
    ) as cursor:
        pass
    
    # Query with named parameters
    async with db.execute(
        "SELECT * FROM users WHERE name = :name",
        {"name": "John Doe"}
    ) as cursor:
        async for row in cursor:
            print(row)

Batch Query Execution

Execute the same SQL statement multiple times with different parameter sets efficiently.

async def executemany(
    self,
    sql: str,
    parameters: Iterable[Iterable[Any]]
) -> Cursor:
    """
    Helper to create a cursor and execute the given multiquery.
    
    Parameters:
    - sql: SQL statement to execute multiple times
    - parameters: Iterable of parameter tuples, one for each execution
    
    Returns:
    Cursor: Async cursor after batch execution
    """

Usage example:

async with aiosqlite.connect("database.db") as db:
    # Insert multiple records efficiently
    users_data = [
        (1, "Alice", "alice@example.com"),
        (2, "Bob", "bob@example.com"),
        (3, "Charlie", "charlie@example.com")
    ]
    
    async with db.executemany(
        "INSERT INTO users (id, name, email) VALUES (?, ?, ?)",
        users_data
    ) as cursor:
        pass
    
    await db.commit()

Script Execution

Execute multiple SQL statements from a script string, useful for database initialization and migrations.

async def executescript(self, sql_script: str) -> Cursor:
    """
    Helper to create a cursor and execute a user script.
    
    Parameters:
    - sql_script: String containing multiple SQL statements separated by semicolons
    
    Returns:
    Cursor: Async cursor after script execution
    """

Usage example:

async with aiosqlite.connect("database.db") as db:
    schema_script = """
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY,
        name TEXT NOT NULL,
        email TEXT UNIQUE
    );
    
    CREATE TABLE IF NOT EXISTS posts (
        id INTEGER PRIMARY KEY,
        user_id INTEGER,
        title TEXT,
        content TEXT,
        FOREIGN KEY (user_id) REFERENCES users (id)
    );
    
    INSERT OR IGNORE INTO users (name, email) 
    VALUES ('Admin', 'admin@example.com');
    """
    
    async with db.executescript(schema_script) as cursor:
        pass

Specialized Query Methods

Convenience methods for common query patterns with optimized execution.

async def execute_insert(
    self,
    sql: str,
    parameters: Optional[Iterable[Any]] = None
) -> Optional[sqlite3.Row]:
    """
    Helper to insert and get the last_insert_rowid.
    
    Parameters:
    - sql: INSERT statement to execute
    - parameters: Optional parameter values for query placeholders
    
    Returns:
    Optional[sqlite3.Row]: Row containing the last_insert_rowid, or None
    """

async def execute_fetchall(
    self,
    sql: str,
    parameters: Optional[Iterable[Any]] = None
) -> Iterable[sqlite3.Row]:
    """
    Helper to execute a query and return all the data.
    
    Parameters:
    - sql: SELECT statement to execute
    - parameters: Optional parameter values for query placeholders
    
    Returns:
    Iterable[sqlite3.Row]: All result rows
    """

Usage example:

async with aiosqlite.connect("database.db") as db:
    # Insert and get the new row ID (using context manager)
    async with db.execute_insert(
        "INSERT INTO users (name, email) VALUES (?, ?)",
        ("Jane Doe", "jane@example.com")
    ) as new_row:
        if new_row:
            print(f"New user ID: {new_row[0]}")
    
    # Execute query and get all results at once (using context manager)
    async with db.execute_fetchall("SELECT * FROM users") as all_users:
        for user in all_users:
            print(f"User: {user}")

Cursor Management

Direct cursor creation and management for advanced query operations.

async def cursor(self) -> Cursor:
    """
    Create an aiosqlite cursor wrapping a sqlite3 cursor object.
    
    Returns:
    Cursor: New async cursor instance
    """

Cursor Class Methods

The Cursor class provides fine-grained control over query execution and result fetching.

class Cursor:
    """Async SQLite database cursor for query execution and result iteration."""
    
    def __init__(self, conn: Connection, cursor: sqlite3.Cursor) -> None: ...
    
    async def __aenter__(self) -> "Cursor":
        """Async context manager entry."""
        
    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
        """Async context manager exit with automatic cleanup."""
    
    def __aiter__(self) -> AsyncIterator[sqlite3.Row]:
        """Enable async iteration over result rows."""

Cursor Query Execution

Execute queries directly on cursor instances.

async def execute(
    self,
    sql: str,
    parameters: Optional[Iterable[Any]] = None
) -> "Cursor":
    """
    Execute the given query.
    
    Parameters:
    - sql: SQL statement to execute
    - parameters: Optional parameter values for query placeholders
    
    Returns:
    Cursor: Self for method chaining
    """

async def executemany(
    self,
    sql: str,
    parameters: Iterable[Iterable[Any]]
) -> "Cursor":
    """
    Execute the given multiquery.
    
    Parameters:
    - sql: SQL statement to execute multiple times
    - parameters: Iterable of parameter tuples
    
    Returns:
    Cursor: Self for method chaining
    """

async def executescript(self, sql_script: str) -> "Cursor":
    """
    Execute a user script.
    
    Parameters:
    - sql_script: String containing multiple SQL statements
    
    Returns:
    Cursor: Self for method chaining
    """

Data Retrieval

Methods for fetching query results in different patterns.

async def fetchone(self) -> Optional[sqlite3.Row]:
    """
    Fetch a single row.
    
    Returns:
    Optional[sqlite3.Row]: Next result row or None if no more rows
    """

async def fetchmany(self, size: Optional[int] = None) -> Iterable[sqlite3.Row]:
    """
    Fetch up to cursor.arraysize number of rows.
    
    Parameters:
    - size: Maximum number of rows to fetch (defaults to cursor.arraysize)
    
    Returns:
    Iterable[sqlite3.Row]: List of result rows
    """

async def fetchall(self) -> Iterable[sqlite3.Row]:
    """
    Fetch all remaining rows.
    
    Returns:
    Iterable[sqlite3.Row]: List of all remaining result rows
    """

async def close(self) -> None:
    """
    Close the cursor.
    
    Releases cursor resources. Cursor cannot be used after closing.
    """

Cursor Properties

Properties providing access to cursor state and metadata.

@property
def rowcount(self) -> int:
    """
    Number of rows affected by the last operation.
    
    Returns:
    int: Row count (-1 for SELECT statements in some cases)
    """

@property
def lastrowid(self) -> Optional[int]:
    """
    Row ID of the last inserted row.
    
    Returns:
    Optional[int]: Last inserted row ID or None
    """

@property
def arraysize(self) -> int:
    """
    Number of rows to fetch at a time with fetchmany().
    
    Returns:
    int: Default fetch size
    """

@arraysize.setter
def arraysize(self, value: int) -> None:
    """
    Set number of rows to fetch at a time.
    
    Parameters:
    - value: New array size value
    """

@property
def description(self) -> tuple[tuple[str, None, None, None, None, None, None], ...]:
    """
    Column description for the current result set.
    
    Returns:
    tuple: Tuple of column descriptors (name, type_code, display_size, 
           internal_size, precision, scale, null_ok)
    """

@property
def row_factory(self) -> Optional[Callable[[sqlite3.Cursor, sqlite3.Row], object]]:
    """
    Current row factory for this cursor.
    
    Returns:
    Optional[Callable]: Row factory function or None
    """

@row_factory.setter
def row_factory(self, factory: Optional[type]) -> None:
    """
    Set row factory for this cursor.
    
    Parameters:
    - factory: Row factory class or None
    """

@property
def connection(self) -> sqlite3.Connection:
    """
    Reference to the underlying sqlite3 connection.
    
    Returns:
    sqlite3.Connection: The wrapped sqlite3 connection object
    """

Usage example:

async with aiosqlite.connect("database.db") as db:
    async with db.cursor() as cursor:
        await cursor.execute("SELECT id, name, email FROM users WHERE id > ?", (10,))
        
        # Check result metadata
        print(f"Columns: {[desc[0] for desc in cursor.description]}")
        
        # Fetch results in different ways
        first_row = await cursor.fetchone()
        if first_row:
            print(f"First user: {first_row}")
        
        next_batch = await cursor.fetchmany(5)
        print(f"Next {len(next_batch)} users")
        
        # Or iterate through all results
        await cursor.execute("SELECT * FROM users")
        async for row in cursor:
            print(f"User: {row}")

Install with Tessl CLI