136 lines
3.9 KiB
Python
136 lines
3.9 KiB
Python
"""Database switching endpoints — list, switch, create, upload, and download."""
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, UploadFile, File
|
|
from fastapi.responses import FileResponse
|
|
from pydantic import BaseModel
|
|
|
|
from ..auth import verify_token
|
|
from .. import database as db_module
|
|
|
|
router = APIRouter(
|
|
prefix="/api/databases",
|
|
tags=["databases"],
|
|
dependencies=[Depends(verify_token)],
|
|
)
|
|
|
|
|
|
class DatabaseInfo(BaseModel):
|
|
name: str
|
|
path: str
|
|
size_bytes: int
|
|
active: bool
|
|
|
|
|
|
class SwitchRequest(BaseModel):
|
|
path: str
|
|
|
|
|
|
class CreateRequest(BaseModel):
|
|
name: str
|
|
|
|
|
|
class CopyRequest(BaseModel):
|
|
name: str
|
|
|
|
|
|
def _db_info(path: Path, active: bool) -> DatabaseInfo:
|
|
return DatabaseInfo(
|
|
name=path.stem,
|
|
path=str(path),
|
|
size_bytes=path.stat().st_size,
|
|
active=active,
|
|
)
|
|
|
|
|
|
@router.get("", response_model=list[DatabaseInfo])
|
|
def list_databases():
|
|
"""List all .db files in the data directory."""
|
|
current = Path(db_module.DB_PATH).resolve()
|
|
dbs = []
|
|
for f in sorted(db_module.DATA_DIR.glob("*.db")):
|
|
dbs.append(_db_info(f, f.resolve() == current))
|
|
return dbs
|
|
|
|
|
|
@router.post("/switch", response_model=DatabaseInfo)
|
|
def switch_db(req: SwitchRequest):
|
|
"""Switch to a different database file. Path must be an existing .db file."""
|
|
db_path = Path(req.path)
|
|
if not db_path.exists():
|
|
raise HTTPException(404, f"Database not found: {req.path}")
|
|
if db_path.suffix != ".db":
|
|
raise HTTPException(400, "Path must point to a .db file")
|
|
db_module.switch_database(db_path)
|
|
return _db_info(db_path, True)
|
|
|
|
|
|
@router.post("/create", response_model=DatabaseInfo)
|
|
def create_db(req: CreateRequest):
|
|
"""Create a new empty database and switch to it."""
|
|
name = req.name.strip()
|
|
if not name:
|
|
raise HTTPException(400, "Name is required")
|
|
if not name.endswith(".db"):
|
|
name += ".db"
|
|
db_path = db_module.DATA_DIR / name
|
|
if db_path.exists():
|
|
raise HTTPException(409, f"Database already exists: {name}")
|
|
db_module.switch_database(db_path)
|
|
return _db_info(db_path, True)
|
|
|
|
|
|
@router.post("/copy", response_model=DatabaseInfo)
|
|
def copy_current_db(req: CopyRequest):
|
|
"""Copy the current active database to a new file (snapshot/backup)."""
|
|
name = req.name.strip()
|
|
if not name:
|
|
raise HTTPException(400, "Name is required")
|
|
if not name.endswith(".db"):
|
|
name += ".db"
|
|
dest = db_module.DATA_DIR / name
|
|
if dest.exists():
|
|
raise HTTPException(409, f"Database already exists: {name}")
|
|
shutil.copy2(db_module.DB_PATH, dest)
|
|
return _db_info(dest, False)
|
|
|
|
|
|
@router.get("/active", response_model=DatabaseInfo)
|
|
def get_active():
|
|
"""Return info about the currently active database."""
|
|
return _db_info(Path(db_module.DB_PATH), True)
|
|
|
|
|
|
@router.post("/upload", response_model=DatabaseInfo)
|
|
def upload_db(file: UploadFile = File(...), switch: bool = True):
|
|
"""Upload a .db file from the client and optionally switch to it."""
|
|
if not file.filename or not file.filename.endswith(".db"):
|
|
raise HTTPException(400, "File must be a .db file")
|
|
# Sanitize filename
|
|
safe_name = Path(file.filename).name
|
|
dest = db_module.DATA_DIR / safe_name
|
|
# If it already exists, add a suffix
|
|
counter = 1
|
|
while dest.exists():
|
|
dest = db_module.DATA_DIR / f"{Path(safe_name).stem}_{counter}.db"
|
|
counter += 1
|
|
with open(dest, "wb") as f:
|
|
shutil.copyfileobj(file.file, f)
|
|
if switch:
|
|
db_module.switch_database(dest)
|
|
return _db_info(dest, switch)
|
|
|
|
|
|
@router.get("/download")
|
|
def download_db():
|
|
"""Download the currently active database file."""
|
|
p = Path(db_module.DB_PATH)
|
|
if not p.exists():
|
|
raise HTTPException(404, "Active database not found")
|
|
return FileResponse(
|
|
path=str(p),
|
|
filename=p.name,
|
|
media_type="application/octet-stream",
|
|
)
|