| from collections.abc import Generator |
|
|
| from sqlalchemy import create_engine |
| from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker |
|
|
| from app.core.config import get_settings |
|
|
|
|
| class Base(DeclarativeBase): |
| pass |
|
|
|
|
| settings = get_settings() |
| connect_args = {"check_same_thread": False} if settings.database_url.startswith("sqlite") else {} |
| engine = create_engine(settings.database_url, pool_pre_ping=True, connect_args=connect_args) |
| SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False) |
|
|
|
|
| def get_db() -> Generator[Session, None, None]: |
| db = SessionLocal() |
| try: |
| yield db |
| finally: |
| db.close() |
|
|
|
|
