File size: 1,862 Bytes
6b46266 78e5f5e 6b46266 abb7cb4 30b2253 78e5f5e 30b2253 abb7cb4 30b2253 abb7cb4 30b2253 6b46266 30b2253 78e5f5e 6b46266 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 |
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)
|