or run

npx @tessl/cli init

Version

Tile

Overview

Evals

Files

docs

advanced.md connection.md index.md queries.md transactions.md

tile.json

tessl/pypi-aiosqlite

asyncio bridge to the standard sqlite3 module

Workspace: tessl
Visibility: Public
Created: 3 months ago
Last updated: 3 months ago
Describes: pkg:pypi/aiosqlite@0.21.x

To install, run

npx @tessl/cli install tessl/pypi-aiosqlite@0.21.0

aiosqlite

Asyncio bridge to the standard sqlite3 module. Provides async/await compatible interface for SQLite database operations while maintaining compatibility with the standard sqlite3 module API patterns and enabling non-blocking database operations in async applications.

Package Information

Package Name: aiosqlite
Package Type: pypi
Language: Python
Installation: pip install aiosqlite

Core Imports

import aiosqlite

For specific components:

from aiosqlite import connect, Connection, Cursor

Basic Usage

import aiosqlite
import asyncio

async def main():
    # Connect using context manager (recommended)
    async with aiosqlite.connect("example.db") as db:
        # Create table
        await db.execute('''
            CREATE TABLE IF NOT EXISTS users (
                id INTEGER PRIMARY KEY,
                name TEXT NOT NULL,
                email TEXT
            )
        ''')
        
        # Insert data
        await db.execute(
            "INSERT INTO users (name, email) VALUES (?, ?)",
            ("John Doe", "john@example.com")
        )
        
        # Commit transaction
        await db.commit()
        
        # Query data using cursor
        async with db.execute("SELECT * FROM users") as cursor:
            async for row in cursor:
                print(f"User: {row[1]}, Email: {row[2]}")

# Run the async function
asyncio.run(main())

Architecture

The library implements a thread-based architecture to prevent blocking the event loop:

Connection Thread: Each connection runs in a separate thread, executing SQLite operations
Queue-based Communication: Async methods queue operations and await results through futures
Context Managers: Automatic resource cleanup for connections and cursors
sqlite3 Compatibility: Full API compatibility with the standard sqlite3 module

Capabilities

Database Connection Management

Core connection functionality including connection creation, transaction management, and resource cleanup. Provides async context managers for automatic resource management.

def connect(database: Union[str, Path], *, iter_chunk_size=64, **kwargs) -> Connection: ...

Connection Management

Query Execution and Data Retrieval

Comprehensive query execution methods including single queries, batch operations, and data fetching patterns. Supports all standard SQL operations with async/await syntax.

async def execute(self, sql: str, parameters: Optional[Iterable[Any]] = None) -> Cursor: ...
async def executemany(self, sql: str, parameters: Iterable[Iterable[Any]]) -> Cursor: ...
async def fetchone(self) -> Optional[sqlite3.Row]: ...
async def fetchall(self) -> Iterable[sqlite3.Row]: ...

Query Operations

Transaction and State Management

Transaction control, isolation levels, and connection state management. Includes commit, rollback operations and transaction properties.

async def commit(self) -> None: ...
async def rollback(self) -> None: ...
@property
def in_transaction(self) -> bool: ...
@property  
def isolation_level(self) -> Optional[str]: ...

Transactions

Advanced Database Features

Extended SQLite functionality including user-defined functions, database backups, progress handlers, and extension loading.

async def create_function(self, name: str, num_params: int, func: Callable, deterministic: bool = False) -> None: ...
async def backup(self, target: Union[Connection, sqlite3.Connection], **kwargs) -> None: ...
async def iterdump(self) -> AsyncIterator[str]: ...

Advanced Features

Constants and Utilities

Package metadata and SQLite configuration constants re-exported from the sqlite3 module.

__version__: str  # Package version string (e.g., "0.21.0")
paramstyle: str  # Parameter style supported by sqlite3 ("qmark")
sqlite_version: str  # SQLite library version string
sqlite_version_info: tuple[int, int, int]  # SQLite version as tuple (major, minor, micro)

Type Adapters and Converters

Functions for registering custom type handling between Python and SQLite.

def register_adapter(type_: type, adapter: Callable) -> None:
    """
    Register an adapter callable for a Python type.
    
    Parameters:
    - type_: Python type to register adapter for
    - adapter: Function to convert Python type to SQLite-compatible type
    """

def register_converter(typename: str, converter: Callable) -> None:
    """
    Register a converter callable for a database type.
    
    Parameters:
    - typename: SQLite type name to register converter for  
    - converter: Function to convert SQLite type to Python type
    """

Usage example:

import aiosqlite
import json
from datetime import datetime

# Register JSON adapter/converter
def adapt_json(obj):
    return json.dumps(obj).encode('utf-8')

def convert_json(s):
    return json.loads(s.decode('utf-8'))

aiosqlite.register_adapter(dict, adapt_json)
aiosqlite.register_converter("json", convert_json)

# Register datetime adapter/converter  
def adapt_datetime(dt):
    return dt.isoformat()

def convert_datetime(s):
    return datetime.fromisoformat(s.decode('utf-8'))

aiosqlite.register_adapter(datetime, adapt_datetime)
aiosqlite.register_converter("timestamp", convert_datetime)

Types

class Connection:
    """Async SQLite database connection."""
    
class Cursor:
    """Async SQLite database cursor."""

class Row:
    """
    Dict-like row object for query results (re-exported from sqlite3).
    
    Provides both index-based and name-based access to column values.
    Available when using sqlite3.Row as row_factory.
    """
    def __getitem__(self, key: Union[int, str]) -> Any: ...
    def keys(self) -> list[str]: ...

# Type aliases
IsolationLevel = Optional[Literal["DEFERRED", "IMMEDIATE", "EXCLUSIVE"]]

Exceptions

All exceptions are re-exported from the sqlite3 module:

# Base exceptions
class Error(Exception): ...
class Warning(Exception): ...
class DatabaseError(Error): ...

# Specific exceptions  
class IntegrityError(DatabaseError): ...
class ProgrammingError(DatabaseError): ...
class OperationalError(DatabaseError): ...
class NotSupportedError(DatabaseError): ...