AquiliaDatabase — Aquilia Documentation
Comprehensive guide and documentation for AquiliaDatabase in the Aquilia framework. View API reference, examples, and implementation patterns.
Docs / Database / AquiliaDatabase AquiliaDatabase The central async database engine — connection lifecycle, raw query API, transactions, savepoints, and introspection. Registered as a DI @service(scope="app"). AquiliaDatabase Source: aquilia/db/engine.py:89 · DI scope: app Constructor class AquiliaDatabase: def __init__( self, url: str | None = None, *, config: DatabaseConfig | None = None, connect_retries: int = 3, connect_retry_delay: float = 0.5, **options, ): ... Param Type Description ))} from aquilia.db import AquiliaDatabase from aquilia.db.configs import PostgresConfig, SqliteConfig # URL string db = AquiliaDatabase("sqlite:///app.db") db = AquiliaDatabase("postgresql://admin:s3cr3t@localhost/mydb") # Typed config (recommended — IDE autocompletion) db = AquiliaDatabase(config=PostgresConfig( host="localhost", database="mydb", user="admin", password="s3cr3t", pool_size=20, sslmode="require", )) # Custom retry db = AquiliaDatabase("postgresql://...", connect_retries=5, connect_retry_delay=1.0) Connection Management Method Description ))} Query Methods Method Returns Description ))} # All methods use ? placeholders — adapter normalises per backend rows = await db.fetch_all("SELECT * FROM users WHERE active = ?", [True]) row = await db.fetch_one("SELECT * FROM users WHERE id = ?", [42]) count = await db.fetch_val("SELECT COUNT(*) FROM orders WHERE status = ?", ["paid"]) res = await db.execute("UPDATE users SET active = ? WHERE id = ?", [False, 99]) print(res.rowcount) # rows affected await db.execute_many( "INSERT INTO tags (name) VALUES (?)", [["python"], ["async"], ["orm"]], ) Transactions Use the transaction() context manager for automatic commit / rollback. For manual control call begin(), commit(), rollback() directly. # Context manager (recommended) async with db.transaction(): await db.execute("INSERT INTO orders ...") await db.execute("UPDATE inventory ...") # Commits if no exception, rolls back otherwise # Manual await db.begin(isolation="REPEATABLE READ") try: await db.execute("UPDATE accounts SET balance = balance - ? WHERE id = ?", [100, 1]) await db.execute("UPDATE accounts SET balance = balance + ? WHERE id = ?", [100, 2]) await db.commit() except Exception: await db.rollback() raise Savepoints Nested checkpoints inside a transaction. Name must match [a-zA-Z_][a-zA-Z0-9_]*. await db.begin() await db.savepoint("before_risky_op") try: await db.execute("DELETE FROM temporary_data WHERE expired = ?", [True]) await db.release_savepoint("before_risky_op") # commit savepoint except Exception: await db.rollback_to_savepoint("before_risky_op") # undo only this block await db.commit() ⚠️ Prefer atomic() from the ORM layer when working with Model.save() — it integrates with signals and on_commit hooks. Use db.transaction() for raw SQL blocks only. Schema Introspection Method Returns Description ))} tables = await db.get_tables() # ["users", "posts", "tags", "migrations"] exists = await db.table_exists("sessions") cols = await db.get_columns("users") for col in cols: print(f" : ") Module-Level Functions from aquilia.db import configure_database, get_database, set_database, get_all_databases # Configure (called automatically by Integration.database()) db = configure_database(url="sqlite:///app.db") db = configure_database(config=PostgresConfig(...), alias="primary") # Retrieve db = get_database() # default db = get_database("primary") # named alias # Replace an existing instance set_database(my_db, alias="default") # All configured databases all_dbs = get_all_databases() # Faults Fault When raised ))} from aquilia.models import DatabaseConnectionFault, QueryFault try: rows = await db.fetch_all("SELECT * FROM users") except QueryFault as e: print(e.reason, e.metadata) except DatabaseConnectionFault as e: print("Connection lost:", e.reason) Overview Config Classes )
Go to Homepage