Database API reference

You can also view the Sphinx docs for the database at https://docs.mau.fi/python/latest/api/mautrix.util/async_db.html, but as of writing, they aren't ready yet.

Database

acquire is an async context manager that returns a Connection:

async with self.database.acquire() as conn:
    conn.execute(...)

The class also contains execute, executemany, fetch, fetchrow, fetchval and table_exists as convenience methods, which simply acquire a connection and run the single method using it (see reference below).

Connection

Records returned by the fetch methods are either asyncpg.Record or the stdlib's sqlite3.Row`. Both work mostly the same way (can access fields by both column index and name).

• execute(query: str, *args: Any) -> str - Execute a query without reading the response:

  await conn.execute("INSERT INTO foo (text) VALUES ($1)", "hello world")
  

• executemany(query: str, *args: Any) -> str - Execute a query multiple times:

  await conn.execute(
    "INSERT INTO foo (text) VALUES ($1)",
    [("hello world 1",), ("hello world 2",), ("hello world 3",)],
  )
  

• fetch(query: str, *args: Any) -> list[Row | Record] - Execute a query and get a list of rows in response:

  rows = await conn.fetch("SELECT text FROM foo")
  for row in rows:
    print(row["text"])
  

• fetchrow(query: str, *args: Any) -> Row | Record | None - Execute a query and get the first rows (or None, if there are no rows):

  row = await conn.fetchrow("SELECT text FROM foo WHERE test=1")
  if row:
    print("Found", row["text"])
  else:
    print("Row not found :(")
  

• fetchval(query: str, *args: Any, column: int = 0) -> Any - Execute a query and get a single column from the first row (or None if there are no rows):

  text = await conn.fetchval("SELECT text FROM foo WHERE test=1")
  print(text)
  

• table_exists(name: str) -> bool - Check if a table exists in the database.
• Postgres only: copy_records_to_table(table_name: str, *, records: list[tuple[Any, ...]], columns: tuple[str, ...]) -> None - Efficiently insert multiple rows into the database:

  await conn.copy_records_to_table(
    table_name="foo",
    records=[("hello world 1",), ("hello world 2",), ("hello world 3",)],
    columns=("text",),
  )