35 lines
1.1 KiB
Python
35 lines
1.1 KiB
Python
import os
|
|
from pathlib import Path
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker, DeclarativeBase
|
|
|
|
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./data/cluedo.db")
|
|
|
|
engine_options = {"pool_pre_ping": True}
|
|
|
|
if DATABASE_URL.startswith("sqlite"):
|
|
# SQLite does not allow connections to be used from another thread by
|
|
# default. FastAPI/SQLAlchemy may use a connection across request threads.
|
|
engine_options["connect_args"] = {"check_same_thread": False}
|
|
|
|
# Make the parent directory available for the default local database.
|
|
if DATABASE_URL.startswith("sqlite:///"):
|
|
sqlite_path = DATABASE_URL.removeprefix("sqlite:///")
|
|
if sqlite_path not in (":memory:", ""):
|
|
Path(sqlite_path).expanduser().parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
engine = create_engine(DATABASE_URL, **engine_options)
|
|
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)
|
|
|
|
class Base(DeclarativeBase):
|
|
pass
|
|
|
|
def get_db():
|
|
db = SessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|