Coverage for volexport/api_mgmt.py: 100%

1from fastapi import APIRouter, Request 

2from fastapi.responses import PlainTextResponse 

3from pathlib import Path 

4import datetime 

5from .config import config 

6from .tgtd import Tgtd 

7from logging import getLogger 

8 

9_log = getLogger(__name__) 

10router = APIRouter() 

11 

12 

13def _backup_file(name) -> Path: 

14 assert "/" not in name 

15 assert not name.startswith(".") 

16 dir = Path(config.BACKUP_DIR) 

17 return (dir / name).with_suffix(".backup") 

18 

19 

20@router.post("/mgmt/backup", description="create tgtd backup") 

21def create_backup() -> dict[str, str]: 

22 basename = datetime.datetime.now().strftime("%Y%m%d-%H%M%S") 

23 outpath = _backup_file(basename) 

24 outpath.write_text(Tgtd().dump()) 

25 _log.info("backup created: %s", outpath) 

26 return {"name": basename} 

27 

28 

29@router.get("/mgmt/backup", description="list tgtd backup files") 

30def list_backup() -> list[dict[str, str]]: 

31 dir = Path(config.BACKUP_DIR) 

32 return [{"name": x.with_suffix("").name} for x in sorted(dir.glob("*.backup"))] 

33 

34 

35@router.delete("/mgmt/backup", description="delete old tgtd backup file") 

36def forget_backup(keep: int = 2) -> dict[str, str]: 

37 dir = Path(config.BACKUP_DIR) 

38 files = sorted(dir.glob("*.backup"), reverse=True) 

39 for f in files[keep:]: 

40 _log.info("delete old backup: %s", f) 

41 f.unlink() 

42 return {"status": "OK"} 

43 

44 

45@router.get("/mgmt/backup/{name}", description="download tgtd backup file") 

46def get_backup(name: str) -> PlainTextResponse: 

47 path = _backup_file(name) 

48 if path.exists(): 

49 return PlainTextResponse(path.read_text()) 

50 raise FileNotFoundError("backup file not found") 

51 

52 

53@router.put("/mgmt/backup/{name}", description="upload tgtd backup file") 

54async def put_backup(name: str, req: Request) -> dict[str, str]: 

55 path = _backup_file(name) 

56 if path.exists(): 

57 raise FileExistsError("backup already exists") 

58 path.write_bytes(await req.body()) 

59 return {"status": "OK"} 

60 

61 

62@router.post("/mgmt/backup/{name}", description="restore tgtd backup") 

63def restore_backup(name: str) -> dict[str, str]: 

64 path = _backup_file(name) 

65 if path.exists(): 

66 Tgtd().restore(path.read_text()) 

67 return {"status": "OK"} 

68 raise FileNotFoundError("backup file not found") 

69 

70 

71@router.delete("/mgmt/backup/{name}", description="delete specified tgtd backup file") 

72def delete_backup(name: str) -> dict[str, str]: 

73 path = _backup_file(name) 

74 if path.exists(): 

75 _log.info("delete backup: %s", path) 

76 path.unlink() 

77 return {"status": "OK"} 

78 raise FileNotFoundError("backup file not found")