DBTx Effect — Aquilia Documentation
Comprehensive guide and documentation for DBTx Effect in the Aquilia framework. View API reference, examples, and implementation patterns.
EFFECTS / DATABASE TRANSACTION Database Transaction Effect The DBTx effect provides transactional database connections from the application pool. Implemented by the DBTxProvider , it automatically commits on request completion or rolls back when faults occur. How Transactions are Bound When a handler requests database access, pulling raw connections manually can lead to leaked sessions or uncommitted transaction blocks. The DBTx effect treats transactions as request-scoped resources. It wraps request processing with an atomic database scope, enforcing database safety at the framework boundaries. Token Configuration & Modes You can declare your database requirements in two modes using Python's class indexing syntax: DBTx["read"] Acquires a database connection configured as read-only. This optimizes SELECT performance, permits routing queries to read replicas, and raises errors if writing is attempted. DBTx["write"] Acquires a database connection and immediately initiates a transaction block. Commits the transaction when the handler finishes successfully, or issues a SQL ROLLBACK if any exception escapes the handler. The DBTxHandle Interface When acquired, the resource injected into the context is an instance of DBTxHandle . It inherits from dict to keep metadata accessible, but exposes async helper wrappers to execute SQL within the active transaction: await handle.execute(sql: str, params: Sequence | None = None) -> AsyncCursor Executes a SQL statement (e.g. INSERT, UPDATE, DELETE) binding parameters to prevent SQL injection. Returns an active query cursor. await handle.fetch_all(sql: str, params: Sequence | None = None) -> list[dict[str, Any]] Executes a query and returns all result rows mapped as key-value dictionaries. await handle.fetch_one(sql: str, params: Sequence | None = None) -> dict[str, Any] | None Fetches a single row. Returns a dictionary or None if no records match. await handle.fetch_val(sql: str, params: Sequence | None = None) -> Any Convenience method that fetches the first column of the first row (ideal for scalar queries like COUNT or returning primary keys). await handle.execute_many(sql: str, params_list: Sequence[Sequence]) -> None Executes the SQL statement iteratively over a list of parameter tuples in a highly optimized batch database operation. Exposed Properties handle.connection — Returns the underlying raw DB engine connection or pool instance. handle.mode — Returns the transaction acquisition mode: "read" or "write". handle.transaction — Points to the active transaction context instance. handle.acquired_at — The monotonic timestamp when the connection was leased. Practical Code Examples 1. Write Operation with Auto-Rollback (E-Commerce Checkout) If any exception is raised during execution, the transaction is rolled back automatically. The example below shows an enterprise e-commerce checkout handler checking inventory, deducting balances, recording audit ledgers, and allocating reward points: from aquilia.controller import Controller, POST, RequestCtx from aquilia.flow import requires from aquilia.effects import DBTx from aquilia.faults.domains import OutOfStockFault, PaymentFailedFault class CheckoutController(Controller): @POST("/checkout") @requires(DBTx["write"]) async def process_checkout(self, ctx: RequestCtx) -> dict: body = await ctx.json() db = ctx.get_effect("DBTx") # DBTxHandle instance # 1. Lock and inspect product stock stock = await db.fetch_val( "SELECT stock FROM products WHERE id = ? FOR UPDATE", (body["product_id"],) ) if stock is None or stock 2. Read-Only Retrieval Optimization (SaaS Tenant Analytics) Using the read-only mode avoids starting transactional locks on the database and signals to the loader that the connection can safely target replica instances: Overview Cache Effect )
Go to Homepage