Connecting to a database¶

Databasez handles database connection pooling and transaction management with minimal fuss. It automatically deals with acquiring and releasing connections and supports transaction APIs for both common and advanced flows.

Database¶

This is the main object used for connections and queries.

from databasez import Database

Constructor parameters¶

url: connection string, DatabaseURL, SQLAlchemy URL, another Database, or None.
force_rollback: enables global rollback mode. Can be overwritten at runtime.
full_isolation: runs the global rollback connection on a dedicated thread/loop.
poll_interval: cross-loop poll interval for multiloop helpers.
config: dictionary-style configuration alternative to url.
**options: forwarded backend/engine options.

Warning

Use either url or config, not both.

`config` shape¶

"connection": {
    "credentials": {
        "scheme": "postgresql+asyncpg",
        "host": "localhost",
        "port": 5432,
        "user": "postgres",
        "password": "password",
        "database": "my_db",
        "options": {
            "ssl": "true"
        }
    }
}

Example:

from databasez import Database

CONFIG = {
    "connection": {
        "credentials": {
            "scheme": "postgresql+asyncpg",
            "host": "localhost",
            "port": 5432,
            "user": "postgres",
            "password": "password",
            "database": "my_db",
            "options": {"ssl": "true", "pool_size": "5"},
        }
    }
}

database = Database(config=CONFIG)

MSSQL-style example:

from databasez import Database

CONFIG = {
    "connection": {
        "credentials": {
            "scheme": "mssql+aioodbc",
            "host": "localhost",
            "port": 1433,
            "user": "sa",
            "password": "Mssql123mssql",
            "database": "master",
            "options": {
                "driver": "ODBC Driver 18 for SQL Server",
                "TrustServerCertificate": "yes",
                "Encrypt": "Optional",
            },
        }
    }
}

database = Database(config=CONFIG)

Main attributes and helpers¶

force_rollback: Context-aware descriptor. Supports assignment/reset and context-manager usage.
engine: Exposes SQLAlchemy AsyncEngine when connected.
asgi(...): ASGI lifespan wrapper for automatic connect/disconnect.

Connecting and disconnecting¶

Use async context manager:

async with Database(DATABASE_URL) as database:
    await database.fetch_val("SELECT 1")

Or explicit lifecycle calls:

database = Database(DATABASE_URL)
await database.connect()
await database.fetch_val("SELECT 1")
await database.disconnect()

Query shortcuts on `Database`¶

Database exposes connection-backed shortcuts:

fetch_all
fetch_one
fetch_val
execute
execute_many
iterate
batched_iterate
run_sync
create_all
drop_all

For detailed query examples see Queries.

Connection options¶

The SQLAlchemy backend extracts some URL query options automatically:

ssl
echo
echo_pool
pool_size
max_overflow
pool_recycle
isolation_level

Example:

database = Database("postgresql+asyncpg://localhost/example?ssl=true&pool_size=20")

Keyword options can also be passed directly:

database = Database("postgresql+asyncpg://localhost/example", ssl=True, pool_size=20)

Force rollback¶

Force rollback can be set globally or temporarily:

database = Database("sqlite+aiosqlite:///example.db", force_rollback=True)

with database.force_rollback(False):
    # next connection() calls in this context are not forced to rollback
    pass

For a runnable example see:

from databasez import Database

setup_database = Database("sqlite+aiosqlite:///testsuite.sqlite3")
database = Database("sqlite+aiosqlite:///testsuite.sqlite3", force_rollback=True)


async def test_foo() -> None:
    async with setup_database:
        await setup_database.execute(
            "CREATE TABLE IF NOT EXISTS notes (id INTEGER PRIMARY KEY, text VARCHAR(100))"
        )

    async with database:
        await database.execute("DELETE FROM notes")
        await database.execute("INSERT INTO notes(text) VALUES (:text)", {"text": "inside-test"})
        row = await database.fetch_val("SELECT COUNT(*) FROM notes")
        assert row == 1

    async with database:
        row = await database.fetch_val("SELECT COUNT(*) FROM notes")
        assert row == 0

ASGI integration helper¶

Database.asgi() wraps an ASGI app and hooks startup/shutdown lifecycle events.

from django.core.asgi import get_asgi_application

from databasez import Database

database = Database("sqlite+aiosqlite:///foo.sqlite")

# `handle_lifespan=True` allows integration with frameworks that don't
# expose ASGI lifespan events themselves.
application = database.asgi(get_asgi_application(), handle_lifespan=True)

Reusing the SQLAlchemy engine¶

If a database is connected, you can access database.engine and use SQLAlchemy APIs directly.

Debugging multiloop behavior¶

To debug blocked cross-loop waits, set:

databasez.utils.DATABASEZ_RESULT_TIMEOUT = <seconds>

This raises instead of waiting forever.

For deeper internals: