Перейти к содержанию

Асинхронная ORM

EndoCore работает под ASGI, где блокировка event loop вредит всем. ORM предоставляет async API — по умолчанию он запускает (проверенную боем) синхронную ORM в тредпуле через asyncio.to_thread, так что обработчики остаются неблокирующими и на SQLite, и на PostgreSQL. На PostgreSQL можно вместо этого включить полностью нативный async-путь — см. раздел «Нативный async на PostgreSQL» ниже.

Почему по умолчанию тредпул, а не нативные async-драйверы

Выгрузка в тредпул даёт неблокирующий доступ поверх ровно того же движка запросов, для обоих диалектов, без дублирования кода — и именно это использует каждый async-метод ниже, если вы не включите async_native=True. У каждого алиаса — небольшой пул соединений (configure(..., pool_size=N); SQLite 1, PostgreSQL 5), а SQLite-соединения открываются с check_same_thread=False, так что использование из разных потоков безопасно.

Manager и QuerySet

У каждой распространённой терминальной операции есть async-близнец с префиксом a:

# создание / чтение
user = await User.objects.acreate(name="Ada", age=36)
user = await User.objects.aget(name="Ada")
n    = await User.objects.acount()
ok   = await User.objects.filter(active=True).aexists()
first = await User.objects.order_by("age").afirst()
last  = await User.objects.order_by("age").alast()
rows  = await User.objects.filter(age__gte=18).alist()

# асинхронная итерация
async for user in User.objects.order_by("name"):
    ...

# запись
await User.objects.all().aupdate(active=True)
await User.objects.filter(spam=True).adelete()
await User.objects.abulk_create([User(name="a"), User(name="b")])
await User.objects.abulk_update(users, ["age"])

# хелперы
user, created = await User.objects.aget_or_create(name="Ada", defaults={"age": 36})
user, created = await User.objects.aupdate_or_create(name="Ada", defaults={"age": 37})
by_pk = await User.objects.ain_bulk([1, 2, 3])
stats = await User.objects.aaggregate(total=Count("*"))

Экземпляры

user = await User.objects.aget(pk=1)
user.age += 1
await user.asave()
await user.arefresh_from_db()
await user.adelete()

Что осталось синхронным

Не все операции ORM имеют async-аналог:

  • Построение QuerySet — filter(), exclude(), order_by(), values(), … никогда не обращается к базе данных, поэтому делать из этого async нечего; базу данных трогает только терминальный вызов, который реально вычисляет запрос (aget, alist, …).
  • ManyRelatedManager — у .add()/.remove()/.set()/.clear() на ManyToManyField пока нет async-формы, ни синхронной, ни нативной. Выгружайте их так же, как и любой другой синхронный вызов из асинхронного кода: await asyncio.to_thread(book.tags.add, tag).

В обработчике

from endocore import Request, Response
from Models.blog import Post

async def handler(request: Request) -> Response:      # Api/v1/Post/Get.py
    posts = await Post.objects.order_by("-id").alist()
    return Response.json({"posts": [p.title for p in posts]})

Транзакции

Используйте async with aatomic(): — a*-вызовы внутри блока присоединяются к транзакции (владение — токен в contextvars, который asyncio.to_thread переносит в тредпул), а соединение из пула берётся вне event loop'а, так что ожидание исчерпанного пула не может заблокировать loop. См. Транзакции.

Замечания

  • Manager даёт async-хелперы чтения/создания; aupdate/adelete живут на QuerySet (вызывайте .all().aupdate(...)) — в соответствии с синхронной семантикой Django.
  • Построение запроса (filter, order_by, …) дёшево и остаётся синхронным; в тредпул выгружается только вычисление.
  • При высокой конкуренции настройте дефолтный тредпул (executor anyio/asyncio) или запустите несколько воркеров.

Нативный async на PostgreSQL

endocore[postgres] уже закрепляет psycopg[binary]>=3.1 — psycopg3 поставляет настоящий async-драйвер (psycopg.AsyncConnection), а не только выгрузку в тредпул выше. Включается явно:

configure(backend="postgres", conninfo="...", async_native=True)

После этого каждый a*-метод у Manager/QuerySet/Model — чтение, запись, bulk-операции, aget_or_create/aupdate_or_create, aaggregate, prefetch, aatomic() — выполняет по-настоящему асинхронную цепочку вызовов без использования рабочего тредпула для операций ORM — на этом пути нет ни одного asyncio.to_thread (у uvicorn, логгера и других executor'ов в процессе всё равно могут быть свои рабочие потоки — речь именно про выполнение запросов самой ORM). Это тот же самый публичный API, что показан выше — в коде обработчика ничего не меняется.

Опция, а не поведение по умолчанию — и только для PostgreSQL

async_native=True никогда не включается сам по себе; поведение уже развёрнутого приложения не меняется, пока оно явно этого не запросит. SQLite остаётся на пути через тредпул, поскольку стандартная библиотека Python не предоставляет для неё совместимого async-драйвера — async_native=True с backend="sqlite" роняет ConfigurationError уже на этапе configure(), а не при первом запросе.

Когда это того стоит

Путь через тредпул уже не блокирует event loop — это не про починку чего-то сломанного. Это важно, когда именно тредпул становится узким местом под реальной конкурентностью: у обычного Python-процесса ограниченное число рабочих потоков по умолчанию, и каждый a*-вызов сейчас занимает один поток на всё время запроса (после завершения запроса поток сразу возвращается в пул). У нативного async такого потолка нет — это по-настоящему одна корутина на каждый запрос в полёте, как и у остальной части ASGI-приложения.

Одно реальное, наблюдаемое отличие в любом случае — отмена задачи. Отмените Task, ожидающую to_thread — и поток продолжит выполнение SQL-запроса до конца, даже если ожидающая его coroutine уже отменена (поток осиротевший, а не остановленный) — отмените Task, ожидающую нативный async-запрос, и сам запрос к базе данных действительно тоже отменяется.

Синхронные вызовы M2M всё ещё нужно выгружать

У ManyRelatedManager (.add()/.remove()/.set()/.clear()) пока вообще нет async API — ни синхронного, ни нативного. Вызов напрямую из корутины, работающей на event loop, сработает, но заблокирует loop на время этого вызова — выгружайте его так же, как и любой другой синхронный вызов из асинхронного кода:

await asyncio.to_thread(book.tags.add, tag)

Закрытие нативных async-соединений при остановке

close_all() закрывает только пул на тредпуле. Если хоть один алиас использует async_native=True, дополнительно дождитесь aclose_all() — обычно из хука on_shutdown:

# hooks.py
from endocore.orm import close_all, aclose_all

async def on_shutdown():
    close_all()
    await aclose_all()

Заметка про Windows

Async-режиму psycopg нужен event loop на основе селектора; политика asyncio по умолчанию на Windows (ProactorEventLoop) не может его выполнить и сразу роняет InterfaceError при подключении. Это касается только локальной разработки на Windows — продакшн ASGI-деплой (uvicorn на Linux/macOS) это не затрагивает. Если нужно погонять async_native=True локально на Windows, запускайте под loop на селекторе:

import asyncio

asyncio.run(main(), loop_factory=asyncio.SelectorEventLoop)  # Python 3.12+
# или на 3.11:
loop = asyncio.SelectorEventLoop()
try:
    loop.run_until_complete(main())
finally:
    loop.close()