| from sqlmodel import SQLModel, create_engine, Session |
| import os |
| from pathlib import Path |
| from dotenv import load_dotenv |
|
|
| load_dotenv() |
|
|
| |
| DEFAULT_URL = "sqlite:////data/app.db" |
| ENV_URL = os.getenv("DATABASE_URL", DEFAULT_URL) |
|
|
| def _sqlite_fs_path(url: str) -> str | None: |
| if not url.startswith("sqlite"): |
| return None |
| if url.startswith("sqlite:////"): |
| return "/" + url.split("sqlite:////", 1)[1] |
| if url.startswith("sqlite:///"): |
| return url.split("sqlite:///", 1)[1] |
| if url.startswith("sqlite://"): |
| return url.split("sqlite://", 1)[1] |
| return None |
|
|
| def _ensure_dir_writable(path: Path) -> bool: |
| try: |
| path.mkdir(parents=True, exist_ok=True) |
| test = path / ".write_test" |
| with open(test, "wb") as f: |
| f.write(b"1") |
| try: |
| test.unlink() |
| except Exception: |
| pass |
| return True |
| except Exception: |
| return False |
|
|
| def _choose_sqlite_url(url: str) -> str: |
| """url の保存先が書き込み不可なら /tmp/app.db に切り替える""" |
| if not url.startswith("sqlite"): |
| return url |
| fs_path = _sqlite_fs_path(url) |
| if fs_path: |
| if _ensure_dir_writable(Path(fs_path).parent): |
| return url |
| |
| fallback = "sqlite:////tmp/app.db" |
| _ensure_dir_writable(Path("/tmp")) |
| print(f"[db] fallback to {fallback} (original: {url})") |
| return fallback |
|
|
| DATABASE_URL = _choose_sqlite_url(ENV_URL) |
|
|
| engine = create_engine( |
| DATABASE_URL, |
| connect_args={"check_same_thread": False} if DATABASE_URL.startswith("sqlite") else {} |
| ) |
|
|
| def init_db(): |
| SQLModel.metadata.create_all(engine) |
|
|
| def get_session(): |
| return Session(engine) |
|
|
