/
nollieundergrob
/
PythonPackageStorage
Обзор
Документация
Войти
/
nollieundergrob
/
PythonPackageStorage
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
Аналитика
Безопасность
master
backend/src/data/db_helper.py
45 строк
1 KB
nollieundergrob
SPLIT
13 мар 2025, 23:11
13 мар 2025, 23:11
ad00f30
Код
Авторство
О чём код?
from sqlalchemy.ext.asyncio import ( AsyncSession, create_async_engine, async_sessionmaker, async_scoped_session, ) from config.settings import settings from asyncio import current_task class AsyncContextException(Exception): def __init__(self, message): self.message = message class DatabaseHelper: def __init__(self, url: str, echo: bool = False): self.engine = create_async_engine(url=url, echo=echo) self.session_factory = async_sessionmaker( bind=self.engine, autocommit=False, expire_on_commit=False ) async def session_dependency(self) -> AsyncSession: async with self.session_factory() as session: yield session async def __aenter__(self) -> AsyncSession: self.session = async_scoped_session( session_factory=self.session_factory, scopefunc=current_task ) return self.session async def __aexit__(self, exc_type, exc_val, exc_tb): if exc_type: raise AsyncContextException( f"Произошла ошибка контекстного менеджера:\n{exc_val}" ) if hasattr(self, 'session'): await self.session.remove() db_helper = DatabaseHelper( url=settings.db_url(), echo=settings.db_echo(), )