Spaces:
Running
Running
| from sqlalchemy import create_engine | |
| from sqlalchemy.orm import DeclarativeBase, sessionmaker | |
| from sqlalchemy.pool import NullPool | |
| from app.core.config import settings | |
| from typing import Any | |
| engine_kwargs: dict[str, Any] = {"pool_pre_ping": True} | |
| if settings.sqlalchemy_database_url.startswith("sqlite"): | |
| engine_kwargs["connect_args"] = {"check_same_thread": False} | |
| else: | |
| db_url = settings.sqlalchemy_database_url | |
| if "6543" in db_url or "pooler.supabase.com" in db_url: | |
| engine_kwargs["poolclass"] = NullPool | |
| else: | |
| engine_kwargs.update( | |
| { | |
| "pool_size": settings.DB_POOL_SIZE, | |
| "max_overflow": settings.DB_MAX_OVERFLOW, | |
| "pool_timeout": 30, | |
| } | |
| ) | |
| engine = create_engine(settings.sqlalchemy_database_url, **engine_kwargs) | |
| SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) | |
| class Base(DeclarativeBase): | |
| pass | |
| def get_db(): | |
| db = SessionLocal() | |
| try: | |
| yield db | |
| finally: | |
| db.close() | |