Raw SQL¶

djox.Database runs SQL from async code, with or without Django.

Without Django¶

"""Quickstart: run SQL from async code with djox."""

import asyncio
import sys
from pathlib import Path

import djox


async def main(path: Path) -> None:
    # One writer connection and a few reader connections, each on its own thread.
    with djox.Database.sqlite(path) as database:
        await database.execute("CREATE TABLE IF NOT EXISTS book (id INTEGER PRIMARY KEY, title TEXT, pages INTEGER)")

        # Writes wait their turn on the writer: no "database is locked" errors.
        await database.execute_many(
            "INSERT INTO book (title, pages) VALUES (%s, %s)",
            [["Dune", 412], ["Emma", 474], ["Beloved", 324]],
        )

        # Reads run in parallel on the readers. `%s` placeholders, like Django.
        long_books, count = await asyncio.gather(
            database.fetch_all("SELECT title FROM book WHERE pages > %s ORDER BY title", [400]),
            database.fetch_one("SELECT count(*) FROM book"),
        )
        print(long_books)  # [('Dune',), ('Emma',)]
        print(count)  # (3,)


if __name__ == "__main__":
    asyncio.run(main(Path(sys.argv[1] if len(sys.argv) > 1 else "quickstart.sqlite3")))

• Open it: djox.Database.sqlite(path, readers=4) (one writer, a few readers) or djox.Database.postgres(url, pool_size=10) (a pool). Use it in a with block, or call close().
• TLS on Postgres: libpq's settings, as arguments: djox.Database.postgres(url, sslmode="verify-full", sslrootcert="ca.pem") (see Postgres).
• Methods: execute(), execute_many(), fetch_all(), fetch_one().
• Placeholders are Django's: %s (%% for a literal %).
• Futures: each method sends the statement right away and returns an asyncio.Future. Await it, or pass several to asyncio.gather().
• Rows are tuples of None, int, float, str and bytes on SQLite, like Python's sqlite3. Parameters can also be bool, datetime, date, time, Decimal and UUID, stored the way Django stores them.

Errors¶

They follow PEP 249, like Python's sqlite3:

The database's errors are subclasses of djox.DatabaseError; all of them, of djox.Error.

Next to Django¶

djox.connections[alias] is the async Database for the same settings as a Django alias, shared by the whole process:

import djox

database = djox.connections["default"]
rows = await database.fetch_all("SELECT title FROM shop_book WHERE pages > %s", [300])

• Same SQL as Django: Django's SQL functions (django_date_extract, REGEXP, STDDEV_POP...) are registered on every connection, so SQL from Django's compiler runs unchanged.
• No hooks, no transaction: it is the layer below the ORM. It doesn't see an open sync transaction (like TestCase's), and hooks don't run.