Coverage for src / server_list / spec / webapi / ups.py: 97%

1#!/usr/bin/env python3 

2""" 

3UPS API. 

4Provides UPS information and topology via REST API from SQLite cache. 

5""" 

6 

7import dataclasses 

8 

9import flask 

10 

11import server_list.spec.data_collector as data_collector 

12import server_list.spec.webapi as webapi 

13 

14ups_api = flask.Blueprint("ups_api", __name__) 

15 

16 

17@ups_api.route("/ups", methods=["GET"]) 

18def get_all_ups(): 

19 """Get all UPS information with topology (clients).""" 

20 ups_info_list = data_collector.get_all_ups_info() 

21 all_clients = data_collector.get_all_ups_clients() 

22 

23 # Group clients by UPS 

24 clients_by_ups: dict[tuple[str, str], list[dict]] = {} 

25 for client in all_clients: 

26 key = (client.ups_name, client.host) 

27 if key not in clients_by_ups: 27 ↛ 29line 27 didn't jump to line 29 because the condition on line 27 was always true

28 clients_by_ups[key] = [] 

29 clients_by_ups[key].append(dataclasses.asdict(client)) 

30 

31 # Build response with topology 

32 result = [] 

33 for ups in ups_info_list: 

34 ups_data = dataclasses.asdict(ups) 

35 key = (ups.ups_name, ups.host) 

36 ups_data["clients"] = clients_by_ups.get(key, []) 

37 result.append(ups_data) 

38 

39 return webapi.success_response(result) 

40 

41 

42@ups_api.route("/ups/<host>/<ups_name>", methods=["GET"]) 

43def get_ups_detail(host: str, ups_name: str): 

44 """Get specific UPS information with clients.""" 

45 ups_info = data_collector.get_ups_info(ups_name, host) 

46 

47 if not ups_info: 

48 return webapi.error_response(f"UPS not found: {ups_name}@{host}", 404) 

49 

50 clients = data_collector.get_ups_clients(ups_name, host) 

51 

52 result = dataclasses.asdict(ups_info) 

53 result["clients"] = [dataclasses.asdict(c) for c in clients] 

54 

55 return webapi.success_response(result)