Frank Schwenk
88bdf6f34b
- Split monolithic app.py into server/api/ routers - Add Pydantic schemas for input validation - Add /agent/* API with separate AGENT_TOKEN auth - Replace static schema.sql with versioned migrations (datenerhaltend) - Add tools/backup_db.py for SQLite backups - Add Dockerfile HEALTHCHECK and compose.yml AGENT_TOKEN env - Add pytest tests for user and agent API - Update README with API docs and deployment safety notes - Bump version to 1.2.0
89 lines
2.6 KiB
Python
Executable File
89 lines
2.6 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Backup the SQLite database used by TakeYourMeds.
|
|
|
|
Usage:
|
|
python tools/backup_db.py [destination_dir]
|
|
|
|
Defaults to ./data/backups/ and keeps the last 14 backups.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import gzip
|
|
import logging
|
|
import pathlib
|
|
import shutil
|
|
import sys
|
|
from datetime import UTC, datetime
|
|
|
|
from server.settings import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
|
|
|
|
DEFAULT_KEEP = 14
|
|
|
|
|
|
def backup_db(destination: pathlib.Path | None = None, keep: int = DEFAULT_KEEP) -> pathlib.Path:
|
|
src = pathlib.Path(settings.db_path)
|
|
if not src.exists():
|
|
raise FileNotFoundError(f"Database not found: {src}")
|
|
|
|
dest_dir = destination or pathlib.Path(src.parent) / "backups"
|
|
dest_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
timestamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ")
|
|
dest_file = dest_dir / f"medis-{timestamp}.sqlite.gz"
|
|
|
|
# SQLite backup via online copy is safe because WAL is not enabled by default.
|
|
# For guaranteed consistency under writes, use the backup API.
|
|
import aiosqlite
|
|
|
|
async def _copy() -> None:
|
|
async with (
|
|
aiosqlite.connect(str(src)) as src_db,
|
|
aiosqlite.connect(str(dest_file).replace(".gz", "")) as dst_db,
|
|
):
|
|
await src_db.backup(dst_db)
|
|
|
|
import asyncio
|
|
|
|
plain = dest_file.with_suffix("")
|
|
asyncio.run(_copy())
|
|
|
|
with open(plain, "rb") as f_in, gzip.open(dest_file, "wb") as f_out:
|
|
shutil.copyfileobj(f_in, f_out)
|
|
plain.unlink()
|
|
|
|
logger.info("Backup created: %s", dest_file)
|
|
|
|
_rotate(dest_dir, keep)
|
|
return dest_file
|
|
|
|
|
|
def _rotate(dest_dir: pathlib.Path, keep: int) -> None:
|
|
backups = sorted(dest_dir.glob("medis-*.sqlite.gz"))
|
|
if len(backups) > keep:
|
|
for old in backups[:-keep]:
|
|
old.unlink()
|
|
logger.info("Removed old backup: %s", old.name)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Backup TakeYourMeds SQLite database")
|
|
parser.add_argument("destination", nargs="?", help="Backup directory (default: ./data/backups)")
|
|
parser.add_argument("--keep", type=int, default=DEFAULT_KEEP, help=f"Number of backups to keep (default: {DEFAULT_KEEP})")
|
|
args = parser.parse_args()
|
|
|
|
dest = pathlib.Path(args.destination) if args.destination else None
|
|
try:
|
|
backup_db(dest, keep=args.keep)
|
|
return 0
|
|
except Exception as exc:
|
|
logger.error("Backup failed: %s", exc)
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|