/
Hetvild
/
RAGFlow
Обзор
Документация
Войти
/
Hetvild
/
RAGFlow
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
dev
src/db/db_helper.py
42 строки
1 KB
bil
feat: инициализация работы с базой + добавлена модель User
24 фев 2026, 11:44
24 фев 2026, 11:44
2e5593f
Код
Авторство
О чём код?
from typing import AsyncGenerator from sqlalchemy.ext.asyncio import ( AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine, ) from core.config import settings class DatabaseHelper: def __init__( self, url: str, echo: bool = False, echo_pool: bool = False, pool_size: int = 5, max_overflow: int = 10, ) -> None: self.engine: AsyncEngine = create_async_engine( url=url, echo=echo, echo_pool=echo_pool, pool_size=pool_size, max_overflow=max_overflow, ) self.session: async_sessionmaker[AsyncSession] = async_sessionmaker( bind=self.engine, autoflush=False, autocommit=False, expire_on_commit=False ) async def dispose(self) -> None: await self.engine.dispose() async def get_session(self) -> AsyncGenerator[AsyncSession, None]: async with self.session() as session: yield session db_helper = DatabaseHelper(url=str(settings.db.DATABASE_URL), echo=True, echo_pool=True)