Async Patterns¶

The entire request path is async, running on ASGI via gunicorn + uvicorn workers. This page documents the async patterns used throughout the codebase and their key constraints.

ASGI Request Path¶

Every layer in the stack uses async def:

Layer	Pattern
Controllers	`async def` methods on django-ninja-extra controllers
Services	`async def` using Django's native async ORM
Selectors	`async def` returning querysets
Providers	`async def` for external API calls

Django Async ORM¶

Services and selectors use Django's native async ORM methods:

Sync	Async
`Model.objects.create()`	`Model.objects.acreate()`
`Model.objects.get()`	`Model.objects.aget()`
`instance.save()`	`instance.asave()`
`instance.delete()`	`instance.adelete()`
`for obj in queryset:`	`async for obj in queryset:`
`queryset.count()`	`await queryset.acount()`
`queryset.exists()`	`await queryset.aexists()`

async def create_borrower(*, first_name: str, last_name: str, **kwargs) -> Borrower:
    borrower = await Borrower.objects.acreate(
        first_name=first_name,
        last_name=last_name,
        **kwargs,
    )
    return borrower

async def get_active_loans(borrower_id: UUID) -> list[Loan]:
    loans = []
    async for loan in Loan.objects.filter(borrower_id=borrower_id, status="active"):
        loans.append(loan)
    return loans

Transaction Handling¶

Django Atomic lacks async context manager support

Django's Atomic class does not have __aenter__/__aexit__ methods. Using async with transaction.atomic() raises TypeError.

Use the synchronous with statement instead --- it works correctly inside async def functions:

async def approve_loan(*, loan_id: UUID) -> Loan:
    # Correct: use sync "with" even in async function
    with transaction.atomic():
        loan = await Loan.objects.aget(id=loan_id)
        loan.status = LoanStatus.APPROVED
        await loan.asave()
    return loan

Do not use select_for_update() with async ORM

select_for_update().aget() fails with TransactionManagementError because the async ORM runs queries in a different thread from the transaction. Remove select_for_update() from async services.

Third-Party Sync Libraries¶

Some third-party libraries don't support async. Use sync_to_async to wrap their calls:

from asgiref.sync import sync_to_async

# django-guardian is sync-only
from guardian.shortcuts import assign_perm

async def assign_loan_permissions(user: User, loan: Loan) -> None:
    await sync_to_async(assign_perm)("view_loan", user, loan)
    await sync_to_async(assign_perm)("change_loan", user, loan)

Only use sync_to_async when necessary --- prefer native async methods when available.

Celery Tasks¶

Celery tasks are always synchronous. They run in their own worker processes, not in the ASGI event loop:

@shared_task(bind=True, base=TenantTask)
def accrue_daily_interest(self: Task[..., Any], loan_id: str) -> None:
    # Use sync ORM directly in task bodies
    loan = Loan.objects.get(id=loan_id)
    # ... perform interest accrual
    loan.save()

Never use async_to_sync in Celery tasks

With CELERY_TASK_ALWAYS_EAGER=True (used in tests and development), async_to_sync in a task body causes RuntimeError: cannot use AsyncToSync in same thread as async event loop. Always use sync ORM directly.

Dispatching Celery Tasks from Async Code¶

Calling .delay() directly from async services can cause issues (segfaults in eager mode due to psycopg C extension). Use the async_task_delay() utility:

from common.tasks import async_task_delay

async def process_payment(payment_id: UUID) -> None:
    # Safe: wraps .delay() in sync_to_async to run in a worker thread
    await async_task_delay(allocate_payment, str(payment_id))

The async_task_delay() function wraps .delay() in sync_to_async() to ensure it runs in a separate thread, avoiding event loop conflicts.

Testing Async Code¶

Async tests require special handling:

import pytest

@pytest.mark.django_db
async def test_create_borrower():
    borrower = await services.create_borrower(
        first_name="Jane",
        last_name="Doe",
    )
    assert borrower.first_name == "Jane"

Note

Test settings include DJANGO_ALLOW_ASYNC_UNSAFE=true because model-bakery's baker.make() uses sync ORM. This allows sync ORM calls within async test functions.