API Reference — Aquilia Documentation
Comprehensive guide and documentation for API Reference in the Aquilia framework. View API reference, examples, and implementation patterns.
SQLite / API Reference SQLite API Reference Detailed interface specifications for SQLite pools, connections, row objects, prepared statement caches, and query methods. Table of Contents , , , , , , ].map((item, i) => ( • ))} create_pool() Initialize a connection pool for the SQLite database. from aquilia.sqlite import create_pool, SqlitePoolConfig async def create_pool( config: SqlitePoolConfig | str, metrics: SqliteMetrics | None = None, ) -> ConnectionPool ConnectionPool Manages reader and writer connections, exposing shortcuts for query execution. class ConnectionPool: def acquire(self, *, readonly: bool = False) -> _AcquireContext: """Acquire a connection from the pool. Returns a context manager yielding AsyncConnection. """ async def close(self) -> None: """Close all connections inside the pool.""" async def execute( self, sql: str, params: Sequence[Any] | None = None, ) -> AsyncCursor: """Helper to acquire writer connection and execute SQL statement.""" async def execute_many( self, sql: str, params_seq: Sequence[Sequence[Any]], ) -> int: """Helper to execute SQL across multiple parameter sets. Returns affected rowcount.""" async def fetch_all( self, sql: str, params: Sequence[Any] | None = None, ) -> list[Row]: """Helper to acquire reader connection and fetch all rows.""" async def fetch_one( self, sql: str, params: Sequence[Any] | None = None, ) -> Row | None: """Helper to acquire reader connection and fetch a single row.""" async def fetch_val( self, sql: str, params: Sequence[Any] | None = None, *, column: int = 0, ) -> Any: """Helper to fetch a single scalar value from the first row.""" AsyncConnection Wraps a single connection, mapping query and transaction calls to thread pool execution. class AsyncConnection: @property def readonly(self) -> bool: """Returns True if this is a read-only reader connection.""" @property def in_transaction(self) -> bool: """Returns True if a transaction is currently active.""" async def execute(self, sql: str, params: Sequence[Any] | None = None) -> AsyncCursor: """Execute a single SQL statement.""" async def execute_many(self, sql: str, params_seq: Sequence[Sequence[Any]]) -> int: """Execute SQL with multiple parameter sets.""" async def fetch_all(self, sql: str, params: Sequence[Any] | None = None) -> list[Row]: """Fetch all rows.""" async def fetch_one(self, sql: str, params: Sequence[Any] | None = None) -> Row | None: """Fetch a single row or None.""" async def fetch_val(self, sql: str, params: Sequence[Any] | None = None, *, column: int = 0) -> Any: """Fetch a single scalar value.""" def transaction(self, *, mode: str = "DEFERRED") -> TransactionContext: """Returns a transaction context manager (DEFERRED, IMMEDIATE, EXCLUSIVE).""" def savepoint_ctx(self, name: str) -> SavepointContext: """Returns a savepoint context manager.""" async def table_exists(self, name: str) -> bool: """Returns True if the table exists.""" async def get_tables(self) -> list[str]: """List all user-defined table names.""" async def backup(self, target: str | AsyncConnection, *, pages: int = -1) -> None: """Perform an online backup to another file or connection.""" Row High-performance, dict-like object wrapping SQLite rows. class Row: def keys(self) -> list[str]: """Get list of column names.""" def values(self) -> list[Any]: """Get list of column values.""" def items(self) -> list[tuple[str, Any]]: """Get list of (column, value) tuples.""" def __getitem__(self, key: int | str) -> Any: """Get column value by integer index or name string.""" def __getattr__(self, name: str) -> Any: """Access column value via attribute name.""" SqlitePoolConfig Configuration dataclass for native SQLite connection pools. @dataclass class SqlitePoolConfig: path: str = "db.sqlite3" journal_mode: str = "WAL" foreign_keys: bool = True busy_timeout: int = 5000 # milliseconds synchronous: str = "NORMAL" cache_size: int = -8000 # negative = KiB mmap_size: int = 268435456 # bytes (256 MB) temp_store: str = "MEMORY" pool_size: int = 5 # reader connections count pool_min_size: int = 2 statement_cache_size: int = 256 query_timeout: float = 30.0 # seconds echo: bool = False auto_commit: bool = True SQLite Error Hierarchy Custom exceptions raised by the native wrapper. Exception Class Base Class Description ))} )
Go to Homepage