Django Async Backend

🚀 Installation & Django Integration

1. Install the package

pip install django-async-backend

2. Django settings

In your settings.py, set the database engine to use the async backend

DATABASES = {
    "default": {
        "ENGINE": "django_async_backend.db.backends.postgresql",
        ...
    },
}

Make sure your Django app (and any other required apps) are listed in INSTALLED_APPS

INSTALLED_APPS = [
    ...
    "django_async_backend",
    ...
]

---

Connection Handler

The connection handler manages database connections for your async backend.

from django_async_backend.db import async_connections

connection = async_connections['default']

async with connection.cursor() as cursor:
    await cursor.execute("SELECT ...")
    rows = await cursor.fetchall()

• Connections are reused and managed automatically.
• Use await connection.close() to manually close a connection if needed.

Independent Connections & Parallel Queries You can create a temporary, isolated connection for parallel or independent operations:

import asyncio
from django_async_backend.db import async_connections

async def run_query():
    async with async_connections.independent_connection():
        conn = async_connections['default']
        async with conn.cursor() as cursor:
            await cursor.execute("SELECT ...")
            return await cursor.fetchall()

results = await asyncio.gather(run_query(), run_query(), run_query())

Cursor

Async cursors provide the following methods:

• execute
• executemany
• fetchone
• fetchmany
• fetchall

async with connection.cursor() as cursor:
    await cursor.execute("SELECT 1")
    row = await cursor.fetchone()

Async Transactions with async_atomic

Basic Usage

Use async_atomic to run async database operations atomically. All changes inside the block are committed together; if an error occurs, all changes are rolled back.

from django_async_backend.db.transaction import async_atomic

async with async_atomic():
    await create_instance(1)
    # If no error, changes are committed
    # If error, changes are rolled back

Rollback on Error

If an exception is raised inside the block, all changes are rolled back:

async with async_atomic():
    await create_instance(1)
    raise Exception("fail")  # Nothing is committed

Nested Transactions (Savepoints)

You can nest async_atomic blocks. Each inner block creates a savepoint. If an error occurs in the inner block, only its changes are rolled back; outer changes remain.

async with async_atomic():
    await create_instance(1)
    try:
        async with async_atomic():
            await create_instance(2)
            raise Exception("fail inner")  # Only instance 2 is rolled back
    except Exception:
        pass
# Only instance 1 is in the database

Using on_commit with async transactions

You can register a callback to run after a successful transaction commit using connection.on_commit.

connection = async_connections[DEFAULT_DB_ALIAS]

async with async_atomic():
    await connection.on_commit(callback)

Writing Async Tests

AsyncioTestCase

Use for async tests that do not require database transactions.

from django_async_backend.test import AsyncioTestCase


class MyAsyncTests(AsyncioTestCase):
    async def asyncSetUp(self):
        # Setup code

    async def asyncTearDown(self):
        # Cleanup code

    async def test_something(self):
        # Your async test logic
        await do_async_stuff()

AsyncioTransactionTestCase

Use for async tests that need database transaction support (rollbacks, atomic blocks).

from django_async_backend.test import AsyncioTransactionTestCase


class MyTransactionTests(AsyncioTransactionTestCase):
    async def asyncSetUp(self):
        # Setup database

    async def asyncTearDown(self):
        # Cleanup database

    async def test_something(self):
        async with async_atomic():
            # DB operations here
            await do_db_stuff()

⚙️ Development Setup

Install pre-commit hooks:

pip install pre-commit
pre-commit install

Install dependencies:

poetry install --with dev

🧪 Running Tests

This project uses a comprehensive test suite powered by unittest.

To run tests:

docker-compose up postgres -d
DJANGO_SETTINGS_MODULE=settings poetry run python -m unittest discover -s tests

Integration tests run locally.

The django_async_backend.db.backends.postgresql backend is fully compatible with Django's default django.db.backends.postgresql backend, as it leverages the default implementation under the hood. To confirm this compatibility, run Django's test suite using the custom backend.

DATABASES = {
    "default": {
        "ENGINE": "django_async_backend.db.backends.postgresql",
        ...
    },
    "other": {
        "ENGINE": "django_async_backend.db.backends.postgresql",
        ...
    },
}

To execute them:

cd tests_django
docker-compose run --build --rm test_django_integration