Database Engine — Aquilia Documentation
Comprehensive guide and documentation for Database Engine in the Aquilia framework. View API reference, examples, and implementation patterns.
Docs / Database Engine Database Engine Aquilia's database layer is a thin, async-first abstraction over multiple backends. One API — SQLite, PostgreSQL, MySQL, or Oracle. All operations use ? placeholders; the adapter normalises to the native param style. How it works AquiliaDatabase is a DI-registered @service(scope="app"). On startup the LifecycleCoordinator calls on_startup(), which opens a connection (with retry logic). All queries go through backend adapters — SQLiteAdapter, PostgresAdapter (asyncpg), MySQLAdapter (aiomysql), OracleAdapter (python-oracledb). The engine records every query duration and emits spans to the active trace for the query inspector. Quick Start Configure the database in workspace.py via the Integration builder: from aquilia.workspace import Workspace, Module from aquilia.integrations import Integration from aquilia.db.configs import PostgresConfig workspace = ( Workspace("my-api") .add_module(Module("users")) .integrate(Integration.database( config=PostgresConfig( host="localhost", database="mydb", user="admin", password="secret", pool_size=10, ) )) .build() ) Or use a URL string: Integration.database(url="postgresql://admin:secret@localhost/mydb") Integration.database(url="sqlite:///app.db") Integration.database(url="mysql://root:pass@localhost/mydb") Supported Backends , , , , ].map(b => ( ))} Raw Queries Use the AquiliaDatabase instance directly for raw SQL. All methods accept ? placeholders. from aquilia.db import get_database db = get_database() # SELECT all rows rows = await db.fetch_all("SELECT * FROM users WHERE active = ?", [True]) # SELECT one row row = await db.fetch_one("SELECT * FROM users WHERE id = ?", [42]) # SELECT scalar value count = await db.fetch_val("SELECT COUNT(*) FROM users") # INSERT / UPDATE / DELETE result = await db.execute("INSERT INTO logs (message) VALUES (?)", ["app started"]) print(result.lastrowid) # Bulk insert await db.execute_many( "INSERT INTO tags (name) VALUES (?)", [["python"], ["async"], ["orm"]], ) DI Injection AquiliaDatabase is registered as a DI service. Inject it into services or controllers with @inject: from aquilia.db import AquiliaDatabase from aquilia.di import inject class UserService: @inject def __init__(self, db: AquiliaDatabase): self.db = db async def get_active_users(self): return await self.db.fetch_all("SELECT * FROM users WHERE active = ?", [True]) Properties Property Type Description ))} Multi-Database Use configure_database() with named aliases to manage multiple connections: from aquilia.db import configure_database, get_database from aquilia.db.configs import PostgresConfig, SqliteConfig # Primary (Postgres) configure_database(config=PostgresConfig(host="pg-host", name="app"), alias="default") # Read replica configure_database(config=PostgresConfig(host="pg-replica", name="app"), alias="replica") # Analytics sidecar configure_database(config=SqliteConfig(path="analytics.db"), alias="analytics") # Retrieve by alias db = get_database() # "default" replica = get_database("replica") stats = get_database("analytics") Engine API )
Go to Homepage