1#!/usr/bin/env python3 

2 

3import flask 

4import json 

5import os 

6from pdu import PDU, PDUPort, PDUState 

7import traceback 

8 

9 

10app = flask.Flask(__name__) 

11 

12 

13@app.before_first_request 

14def populate_static_pdus(): 

15 statics = json.loads(os.environ.get('PDUGATEWAY_STATIC_PDUS', '[]')) 

16 for entry in statics: 

17 if not all([k in entry for k in ['model', 'pdu_name', 'config']]): 

18 raise ValueError("static entry is missing one of 'model', 'pdu_name', or 'config'") 

19 PDU.register(model_name=entry.get('model'), 

20 pdu_name=entry.get('pdu_name'), 

21 config=entry.get('config')) 

22 

23 

24def get_pdu_or_fail(pdu_name): 

25 pdu = PDU.get_by_name(pdu_name) 

26 if pdu is None: 

27 raise ValueError(f"Unknown pdu name '{pdu_name}'") 

28 return pdu 

29 

30 

31def has_no_empty_params(rule): 

32 defaults = rule.defaults if rule.defaults is not None else () 

33 arguments = rule.arguments if rule.arguments is not None else () 

34 return len(defaults) >= len(arguments) 

35 

36 

37@app.errorhandler(ValueError) 

38def handle_valueError_exception(error): 

39 traceback.print_exc() 

40 response = flask.jsonify({"error": str(error)}) 

41 response.status_code = 400 

42 return response 

43 

44 

45@app.route("/") 

46def site_map(): 

47 links = [] 

48 for rule in app.url_map.iter_rules(): 

49 # Filter out rules we can't navigate to in a browser 

50 # and rules that require parameters 

51 if "GET" in rule.methods and has_no_empty_params(rule): 

52 url = flask.url_for(rule.endpoint, **(rule.defaults or {})) 

53 links.append(url) 

54 

55 return {"routes": links} 

56 

57 

58@app.route('/pdu_gateway') 

59def pdu_gateway(): 

60 return { 

61 "version": 1, 

62 "supported_pdus": list(PDU.supported_pdus().keys()), 

63 } 

64 

65 

66@app.route('/v1/pdus', methods=['GET']) 

67def pdus_list(): 

68 return { 

69 "pdus": list(PDU.registered_pdus().keys()) 

70 } 

71 

72 

73@app.route('/v1/pdus', methods=['DELETE']) 

74def delete_all_pdus(): 

75 PDU.full_reset() 

76 return ("", 200) 

77 

78 

79@app.route('/v1/pdus', methods=['POST']) 

80def register_pdu(): 

81 params = flask.request.json 

82 

83 if params is None: 

84 raise ValueError("The request is not JSON-based") 

85 

86 PDU.register(model_name=params.get('model'), 

87 pdu_name=params.get('pdu_name'), 

88 config=params.get('config')) 

89 return ("", 200) 

90 

91 

92# TODO: Allow deleting a single PDU 

93 

94 

95@app.route('/v1/pdus/<pdu_name>/ports') 

96def list_ports(pdu_name): 

97 pdu = get_pdu_or_fail(pdu_name) 

98 

99 ports = {} 

100 for port in pdu.ports: 

101 ports[port.port_id] = { 

102 "state": port.state.name 

103 } 

104 if port.label is not None: 

105 ports[port.port_id]["label"] = port.label 

106 

107 return ports 

108 

109 

110@app.route('/v1/pdus/<pdu_name>/ports/<port_id>/state', methods=['GET']) 

111def get_port_state(pdu_name, port_id): 

112 pdu = get_pdu_or_fail(pdu_name) 

113 return PDUPort(pdu, port_id).state.name + "\n" 

114 

115 

116@app.route('/v1/pdus/<pdu_name>/ports/<port_id>/state', methods=['POST']) 

117def set_port_state(pdu_name, port_id): 

118 pdu = get_pdu_or_fail(pdu_name) 

119 

120 params = flask.request.json 

121 if params is None: 

122 raise ValueError("The request is not JSON-based") 

123 

124 if 'state' not in params: 

125 return ("The request is missing the state that needs to be set", 400) 

126 

127 state = getattr(PDUState, params['state'], None) 

128 if state is None or not state.is_valid_action: 

129 valid_choices = [s.name for s in PDUState.valid_actions()] 

130 return (f"The state '{params['state']}' is invalid: Should be in {valid_choices}\n", 400) 

131 

132 port = PDUPort(pdu, port_id) 

133 port.set(state) 

134 

135 return port.state.name + "\n" 

136 

137 

138# TODO: Add a way to label the ports in the PDU itself 

139 

140if __name__ == '__main__': # pragma: nocover 

141 app.run()