Overview — Aquilia Documentation
Comprehensive guide and documentation for Overview in the Aquilia framework. View API reference, examples, and implementation patterns.
SQLite Module / Overview async SQLite Module Aquilia provides a zero-dependency, native async SQLite connection pool wrapper that makes standard library sqlite3 async-safe by offloading blocking operations to a dedicated thread pool. Quick Start Instantiate the connection pool using create_pool() and a SqlitePoolConfig or a DB URL string. from aquilia.sqlite import create_pool, SqlitePoolConfig # Option A: Simple connection URL (uses WAL mode and default pool settings) pool = await create_pool("sqlite:///app.db") # Option B: Advanced configuration via SqlitePoolConfig config = SqlitePoolConfig( path="app.db", pool_size=10, # Number of concurrent reader connections journal_mode="WAL", # Write-Ahead Logging synchronous="NORMAL", # Optimal WAL speed/durability trade-off ) pool = await create_pool(config) # Quick execution (uses connection pool direct helpers) await pool.execute( "CREATE TABLE IF NOT EXISTS items (id INTEGER PRIMARY KEY, name TEXT)" ) await pool.execute( "INSERT INTO items (name) VALUES (?)", ["Product A"] ) # Fetching rows (returns dict-like Row objects) rows = await pool.fetch_all("SELECT * FROM items") for row in rows: print(f"ID: | Name: ") Workspace Configuration Styles Aquilia support both the legacy static builder integration style and the modern typed-dataclass style inside workspace.py: Modern Style: DatabaseIntegration Dataclass (Recommended) Directly integrate the typed DatabaseIntegration configuration: # workspace.py from aquilia import Workspace from aquilia.integrations import DatabaseIntegration workspace = ( Workspace("myapp") .integrate(DatabaseIntegration( url="sqlite:///app.db", pool_size=8, journal_mode="WAL", synchronous="NORMAL" )) ) Legacy Style: Static Database Builder The legacy static Integration.database() or .database() method: # workspace.py (Legacy) from aquilia import Workspace from aquilia.integrations import Integration workspace = ( Workspace("myapp") .database(url="sqlite:///app.db") ) Warning: The legacy static helper .database() is deprecated and will be removed in a future release. Migrate to direct constructor calls using DatabaseIntegration. Subsystem Features ))} Pool Concurrency Architecture Aquilia's connection pool maintains N reader connections plus exactly 1 writer connection. This matches SQLite's single-writer limitation while allowing concurrent reads in Write-Ahead Log (WAL) mode. CONNECTIONPOOL WAL Concurrency Mode READER POOL N read-only connections Semaphore protected reader_0, reader_1, ... reader_N SERIALIZED WRITER 1 read-write connection asyncio.Lock protected writer_connection (exclusive write) )
Go to Homepage