/
tegdif
/
cism
Обзор
Документация
Войти
/
tegdif
/
cism
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
master
src/core/database.py
36 строк
1 KB
tegdif
first_commit
28 июл 2026, 08:54
28 июл 2026, 08:54
8ecbd07
Код
Авторство
О чём код?
from typing import Optional from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker from sqlalchemy.orm import DeclarativeBase from src.core.config import settings _engine = None def get_engine(): """Возвращает singleton async engine для подключения к БД.""" global _engine if _engine is None: _engine = create_async_engine(settings.database_url, echo=False, pool_size=20, max_overflow=10) return _engine async_session_factory: async_sessionmaker[AsyncSession] = async_sessionmaker( get_engine(), class_=AsyncSession, expire_on_commit=False ) class Base(DeclarativeBase): """Базовый класс для всех SQLAlchemy моделей.""" pass async def get_session() -> AsyncSession: """Предоставляет async сессию БД через FastAPI Depends.""" async with async_session_factory() as session: try: yield session finally: await session.close()