备份数据库
()
| 502 | |
| 503 | @router.post("/database/backup") |
| 504 | async def backup_database(): |
| 505 | """备份数据库""" |
| 506 | import shutil |
| 507 | from datetime import datetime |
| 508 | |
| 509 | settings = get_settings() |
| 510 | |
| 511 | if get_database_backend(settings.database_url) != "sqlite": |
| 512 | raise HTTPException(status_code=400, detail="当前仅支持 SQLite 文件备份,PostgreSQL 请使用 pg_dump") |
| 513 | |
| 514 | db_path = settings.database_url |
| 515 | if db_path.startswith("sqlite:///"): |
| 516 | db_path = db_path[10:] |
| 517 | |
| 518 | if not os.path.exists(db_path): |
| 519 | raise HTTPException(status_code=404, detail="数据库文件不存在") |
| 520 | |
| 521 | # 创建备份目录 |
| 522 | from pathlib import Path as FilePath |
| 523 | backup_dir = FilePath(db_path).parent / "backups" |
| 524 | backup_dir.mkdir(exist_ok=True) |
| 525 | |
| 526 | # 生成备份文件名 |
| 527 | timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") |
| 528 | backup_path = backup_dir / f"database_backup_{timestamp}.db" |
| 529 | |
| 530 | # 复制数据库文件 |
| 531 | shutil.copy2(db_path, backup_path) |
| 532 | |
| 533 | return { |
| 534 | "success": True, |
| 535 | "message": "数据库备份成功", |
| 536 | "backup_path": str(backup_path) |
| 537 | } |
| 538 | |
| 539 | |
| 540 | @router.post("/database/import") |
nothing calls this directly
no test coverage detected