39 lines
1.8 KiB
Python
39 lines
1.8 KiB
Python
"""Асинхронный доступ к БД для Celery-задач: сессия на один вызов задачи.
|
||
|
||
Celery-воркер (`celery ... -B`) работает в prefork-пуле — глобальный async
|
||
engine backend (`core.db.engine`), созданный один раз в родительском
|
||
процессе, не переживает fork (сетевые соединения, открытые до fork, в
|
||
дочернем процессе ведут к зависаниям/битым event loop'ам asyncpg). Поэтому
|
||
каждый вызов задачи создаёт свой временный engine и гарантированно
|
||
уничтожает его по завершении.
|
||
"""
|
||
|
||
import asyncio
|
||
from collections.abc import AsyncGenerator, Callable, Coroutine
|
||
from contextlib import asynccontextmanager
|
||
from typing import Any, TypeVar
|
||
|
||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
||
|
||
from core.config import get_settings
|
||
|
||
T = TypeVar("T")
|
||
|
||
|
||
@asynccontextmanager
|
||
async def open_session() -> AsyncGenerator[AsyncSession, None]:
|
||
"""Открыть сессию БД с короткоживущим engine (закрывается по выходу из контекста)."""
|
||
settings = get_settings()
|
||
engine = create_async_engine(settings.database_url, pool_pre_ping=True)
|
||
session_maker = async_sessionmaker(engine, expire_on_commit=False)
|
||
try:
|
||
async with session_maker() as session:
|
||
yield session
|
||
finally:
|
||
await engine.dispose()
|
||
|
||
|
||
def run_async(factory: Callable[[], Coroutine[Any, Any, T]]) -> T:
|
||
"""Выполнить async-корутину в новом event loop — обёртка для sync Celery-задач."""
|
||
return asyncio.run(factory())
|