Coverage for src / server_list / spec / webapi / storage.py: 100%

1#!/usr/bin/env python3 

2""" 

3Storage API for ZFS pools and mount points. 

4Provides storage information via REST API from SQLite cache. 

5""" 

6 

7import dataclasses 

8from typing import Any 

9 

10import flask 

11 

12import server_list.spec.data_collector as data_collector 

13import server_list.spec.webapi as webapi 

14 

15storage_api = flask.Blueprint("storage_api", __name__) 

16 

17 

18@storage_api.route("/storage/zfs/<host>", methods=["GET"]) 

19def get_host_zfs_pools(host: str): 

20 """Get ZFS pool information for a specific host.""" 

21 pools = data_collector.get_zfs_pool_info(host) 

22 

23 if pools: 

24 return webapi.success_response([dataclasses.asdict(p) for p in pools]) 

25 

26 return webapi.error_response(f"No ZFS pool data for host: {host}") 

27 

28 

29@storage_api.route("/storage/mount/<host>", methods=["GET"]) 

30def get_host_mounts(host: str): 

31 """Get mount point information for a specific host.""" 

32 mounts = data_collector.get_mount_info(host) 

33 

34 if mounts: 

35 return webapi.success_response([dataclasses.asdict(m) for m in mounts]) 

36 

37 return webapi.error_response(f"No mount data for host: {host}") 

38 

39 

40@storage_api.route("/storage/batch", methods=["POST"]) 

41def get_storage_batch(): 

42 """Get storage information for multiple hosts in a single request. 

43 

44 Request body: 

45 { 

46 "zfs_hosts": ["host1", "host2"], # optional 

47 "mount_hosts": ["host3", "host4"] # optional 

48 } 

49 

50 Response: 

51 { 

52 "success": true, 

53 "data": { 

54 "zfs": { 

55 "host1": [pools...], 

56 "host2": [pools...] 

57 }, 

58 "mount": { 

59 "host3": [mounts...], 

60 "host4": [mounts...] 

61 } 

62 } 

63 } 

64 """ 

65 body = flask.request.get_json(silent=True) 

66 if not body or not isinstance(body, dict): 

67 return webapi.error_response("Request body is required", 400) 

68 

69 zfs_hosts: list[str] = body.get("zfs_hosts", []) 

70 mount_hosts: list[str] = body.get("mount_hosts", []) 

71 

72 if not zfs_hosts and not mount_hosts: 

73 return webapi.error_response("At least one of zfs_hosts or mount_hosts is required", 400) 

74 

75 result: dict[str, dict[str, Any]] = {"zfs": {}, "mount": {}} 

76 

77 for host in zfs_hosts: 

78 pools = data_collector.get_zfs_pool_info(host) 

79 result["zfs"][host] = [dataclasses.asdict(p) for p in pools] 

80 

81 for host in mount_hosts: 

82 mounts = data_collector.get_mount_info(host) 

83 result["mount"][host] = [dataclasses.asdict(m) for m in mounts] 

84 

85 return webapi.success_response(result)