src/db.py

from sqlmodel import SQLModel, create_engine, Session
import os
from pathlib import Path
from dotenv import load_dotenv

load_dotenv()

# 既定は /data。開けない場合は /tmp に自動フォールバックします。
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)