Low Level Basics

asyncqlio’s low-level API is a database-agnostic SQL API that provides developers the ability to execute SQL code without worrying about the underlying driver.

from asyncqlio.db import DatabaseInterface

# create the database object to connect to the server.
db = DatabaseInterface("postgresql+asyncpg://joku@127.0.0.1/joku")

async def main():
    # connect to the database with db.connect
    await db.connect()
    # create a transaction to execute sql inside of
    async with db.get_transaction() as trans:
        # run a query
        results: BaseResultSet = await trans.cursor("SELECT 1;")
        row = await results.fetch_row()  # row with 1

Transactions

Transactions are the way of executing queries without affecting the rest of the database. All agnostic connections require the usage of a transaction to execute SQL (it is possible to execute SQL purely on a connection using the driver-specific API, but this is not supported).

The BaseTransaction object is used to abstract away Database API transaction objects into a common format that can be used in every dialect. To get a new transaction that is bound to the current connection, use DatabaseInterface.get_transaction():

# tr is a new transaction object
tr: BaseTransaction = db.get_transaction()
# this is connected to the current database's connections
# and will execute on said connection

Transactions MUST be started before execution can happen; this can be achieved with BaseTransaction.begin().

# start the transaction
# this will usually emit a BEGIN or START TRANSACTION command underneath
await tr.begin()

SQL can be emitted in the transaction with the usage of BaseTransaction.execute() and BaseTransaction.cursor().

# update some data
await tr.execute('UPDATE "user" SET level = 3 WHERE "user".xp < 1000')
# select some rows
rows = await tr.cursor('SELECT * FROM "user" WHERE level > 5')

BaseTransaction.cursor() returns rows from a select query in the form of a BaseResultSet(). ResultSets can be iterated over asynchronously with async for to select each dict-like row:

async for row in rows:
    print(row.keys(), row.values())

Once done with the transaction, you can commit it to flush the changes, or you can rollback to revert any changes.

if all_went_good:
    # it all went good, save changes
    await tr.commit()
else:
    # not all went good, rollback changes
    await tr.rollback()

Transactions support the async for protocol, which will automatically begin and commit/rollback as appropriate.

class asyncqlio.backends.base.BaseTransaction(connector)

Bases: asyncqlio.meta.AsyncABC

The base class for a transaction. This represents a database transaction (i.e SQL statements guarded with a BEGIN and a COMMIT/ROLLBACK).

Children classes must implement:

• BaseTransaction.begin()
• BaseTransaction.rollback()
• BaseTransaction.commit()
• BaseTransaction.execute()
• BaseTransaction.cursor()
• BaseTransaction.close()

Additionally, some extra methods can be implemented:

• BaseTransaction.create_savepoint()
• BaseTransaction.release_savepoint()

These methods are not required to be implemented, but will raise NotImplementedError if they are not.

This class takes one parameter in the constructor: the BaseConnector used to connect to the DB server.

create_savepoint(name)

Creates a savepoint in the current transaction.

Warning

This is not supported in all DB engines. If so, this will raise NotImplementedError.

Parameters:name (str) – The name of the savepoint to create.

release_savepoint(name)

Releases a savepoint in the current transaction.

Parameters:name (str) – The name of the savepoint to release.

coroutine begin()

Begins the transaction, emitting a BEGIN instruction.

coroutine close(self, *, has_error=False)

Called at the end of a transaction to cleanup. The connection will be released if there’s no error; otherwise it will be closed.

Parameters:has_error (bool) – If the transaction has an error.

coroutine commit()

Commits the current transaction, emitting a COMMIT instruction.

coroutine cursor(self, sql, params=None)

Executes SQL and returns a database cursor for the rows.

Parameters:

• sql (str) – The SQL statement to execute.
• params (Optional[Iterable[+T_co]]) – Any parameters to pass to the query.

Return type:

BaseResultSet

Returns:

The BaseResultSet returned from the query, if applicable.

coroutine execute(self, sql, params=None)

Executes SQL in the current transaction.

Parameters:

• sql (str) – The SQL statement to execute.
• params (Optional[Iterable[+T_co]]) – Any parameters to pass to the query.

coroutine rollback(self, checkpoint=None)

Rolls back the transaction.

Parameters:checkpoint (Optional[str]) – If provided, the checkpoint to rollback to. Otherwise, the entire transaction will be rolled back.

class asyncqlio.backends.base.BaseResultSet

Bases: collections.abc.AsyncIterator, asyncqlio.meta.AsyncABC

The base class for a result set. This represents the results from a database query, as an async iterable.

Children classes must implement:

• BaseResultSet.keys
• BaseResultSet.fetch_row
• BaseResultSet.fetch_many

keys

Return type:Iterable[str] Returns:An iterable of keys that this query contained.

coroutine close()

Closes this result set.

coroutine fetch_many(self, n)

Fetches the next N rows in this query.

Parameters:n (int) – The number of rows to fetch. Return type:DictRow

coroutine fetch_row(self)

Fetches the next row in this query.

This should return None if the row could not be fetched.

Return type:DictRow

coroutine flatten(self)

Flattens this ResultSet.

Return type:List[DictRow] Returns:A list of DictRow objects.