/
developer_atatanoff
/
simple_auth_fastapi
Обзор
Документация
Войти
/
developer_atatanoff
/
simple_auth_fastapi
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
master
config/database.py
66 строк
2 KB
Alex Tata
Добавил страницы логирования на вью
03 авг 2026, 15:10
03 авг 2026, 15:10
1002330
Код
Авторство
О чём код?
from contextvars import ContextVar import os from typing import Optional, AsyncGenerator from sqlalchemy.orm import DeclarativeBase from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker from valkey.asyncio import Valkey from config.settings import settings from config.logger import logger DATABASE_URL = settings.DB_LOCAL engine = create_async_engine( DATABASE_URL, echo=False, pool_size=20, max_overflow=40, pool_pre_ping=True, ) class Base(DeclarativeBase): pass AsyncSessionLocal = async_sessionmaker( engine, class_=AsyncSession, expire_on_commit=False, autoflush=False, ) _db_session_context: ContextVar[Optional[AsyncSession]] = ContextVar( 'db_session', default=None ) def get_session_context() -> Optional[AsyncSession]: return _db_session_context.get() def set_session_context(session: AsyncSession): return _db_session_context.set(session) def reset_session_context(token): _db_session_context.reset(token) async def get_db() -> AsyncGenerator[AsyncSession, None]: session = get_session_context() if session is None: raise RuntimeError("Session not found in context. Is middleware registered?") try: yield session except Exception as e: logger.error(f"Database error in route: {e}") raise VALKEY_URL = os.getenv("VALKEY_URL", "redis://localhost:6379") # Создаем ОДИН экземпляр клиента для всего приложения. # decode_responses=True возвращает str вместо bytes valkey_client = Valkey.from_url( VALKEY_URL, decode_responses=True, # Можно настроить параметры пула, если нужно max_connections=20 )