The main Database object. This is the “database interface” to the actual DB server.


DatabaseInterface(dsn: str = None, *, …) The “database interface” to your database.
class asyncqlio.db.DatabaseInterface(dsn: str = None, *, loop: = None)[source]

Bases: object

The “database interface” to your database. This provides the actual connection to the DB server, including things such as querying, inserting, updating, et cetera.

Creating a new database object is simple:

# pass the DSN in the constructor
dsn = "postgresql://postgres:B07_L1v3s_M4tt3r_T00@"
my_database = DatabaseInterface(dsn)
# or provide it in the `.connect()` call
await my_database.connect(dsn)
Parameters:dsn – The Data Source Name <>_ to connect to the database on.
connector = None

The current connector instance.

dialect = None

The current Dialect instance.


Checks if this DB is connected.

bind_tables(md: asyncqlio.orm.schema.table.TableMetadata)[source]

Binds tables to this DB instance.

coroutine connect(dsn: str = None, **kwargs) → asyncqlio.backends.base.BaseConnector[source]

Connects the interface to the database server.


For SQLite3 connections, this will just open the database for reading.

Parameters:dsn – The Data Source Name to connect to, if it was not specified in the constructor.
Returns:The BaseConnector established.
emit_param(name: str) → str[source]

Emits a param in the format that the DB driver specifies.

Parameters:name – The name of the parameter to emit.
Returns:A str representing the emitted param.
get_transaction(**kwargs) → asyncqlio.backends.base.BaseTransaction[source]

Gets a low-level BaseTransaction.

async with db.get_transaction() as transaction:
    results = await transaction.cursor("SELECT 1;")
get_session(**kwargs) → asyncqlio.orm.session.Session[source]

Gets a new Session bound to this instance.

coroutine close()[source]

Closes the current database interface.

coroutine get_db_server_info()[source]

Gets DB server info.


This is not supported on SQLite3 connections.