Coverage for /opt/conda/envs/apienv/lib/python3.10/site-packages/daiquiri/cli/client_requests.py: 0%

1#!/usr/bin/env python 

2# -*- coding: utf-8 -*- 

3 

4"""Example client of the REST server using requests 

5""" 

6 

7import requests 

8import json 

9 

10 

11class RestClient: 

12 base = "http://localhost" 

13 port = 8080 

14 _token = None 

15 

16 def __init__(self, base=None, port=None, verify=True): 

17 if base is not None: 

18 self.base = base 

19 

20 if port is not None: 

21 self.port = port 

22 

23 self.verify = verify 

24 

25 def token(self): 

26 return self._token 

27 

28 def print(self, data): 

29 print(json.dumps(data, indent=4, sort_keys=True)) 

30 

31 def req( 

32 self, 

33 url, 

34 method="get", 

35 data=None, 

36 params=None, 

37 pprint=False, 

38 token=True, 

39 show_headers=False, 

40 ): 

41 parts = self.base.split("/") 

42 base = ( 

43 parts[0] 

44 + "/" 

45 + parts[1] 

46 + "/" 

47 + parts[2] 

48 + ":" 

49 + str(self.port) 

50 + "/" 

51 + "/".join(parts[3:]) 

52 ) 

53 

54 headers = None 

55 if self._token and token: 

56 headers = {"Authorization": "Bearer " + self._token} 

57 resp = getattr(requests, method)( 

58 base + "/api" + url, 

59 json=data, 

60 params=params, 

61 headers=headers, 

62 verify=self.verify, 

63 ) 

64 

65 if pprint: 

66 print(resp.url, resp.status_code, len(resp.content)) 

67 if show_headers: 

68 print(resp.headers) 

69 try: 

70 self.print(resp.json()) 

71 except Exception: 

72 print(resp.text) 

73 

74 return 

75 

76 else: 

77 return resp 

78 

79 def login(self, username, password, client="python-rest"): 

80 auth = {"username": username, "password": password, "client": client} 

81 resp = self.req("/authenticator/login", "post", auth) 

82 if resp.status_code == 201: 

83 self._token = resp.json()["token"] 

84 self._session = resp.json() 

85 else: 

86 try: 

87 print(resp.status_code, resp.json()) 

88 except Exception: 

89 print(resp.status_code, resp.text) 

90 return False 

91 

92 return True 

93 

94 def session(self, key): 

95 if key in self._session: 

96 return self._session[key] 

97 

98 def set_session(self, key, value): 

99 if key in self._session: 

100 self._session[key] = value 

101 

102 def logout(self): 

103 self.req("/authenticator/login", method="delete") 

104 

105 

106if __name__ == "__main__": 

107 client = RestClient() 

108 

109 client.login("abcd", "1234") 

110 client.req( 

111 "/metadata/sessions/select", 

112 method="post", 

113 data={"session": "mt10000-3"}, 

114 pprint=True, 

115 ) 

116 client.req("/session/current/control", method="post", pprint=True) 

117 client.logout()