-
-
Notifications
You must be signed in to change notification settings - Fork 39
/
user.py
134 lines (105 loc) · 4.38 KB
/
user.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
from amqpstorm.compatibility import json
from amqpstorm.compatibility import quote
from amqpstorm.management.base import ManagementHandler
API_USER = 'users/%s'
API_USERS = 'users'
API_USER_VIRTUAL_HOST_PERMISSIONS = 'permissions/%s/%s'
API_USER_PERMISSIONS = 'users/%s/permissions'
class User(ManagementHandler):
def get(self, username):
"""Get User details.
:param str username: Username
:rtype: dict
"""
return self.http_client.get(API_USER % username)
def list(self):
"""List all Users.
:rtype: list
"""
return self.http_client.get(API_USERS)
def create(self, username, password, tags=''):
"""Create User.
:param str username: Username
:param str password: Password
:param str tags: Comma-separate list of tags (e.g. monitoring)
:rtype: None
"""
user_payload = json.dumps({
'password': password,
'tags': tags
})
return self.http_client.put(API_USER % username,
payload=user_payload)
def delete(self, username):
"""Delete User.
:param str username: Username
:rtype: dict
"""
return self.http_client.delete(API_USER % username)
def get_permission(self, username, virtual_host):
"""Get User permissions for the configured virtual host.
:param str username: Username
:param str virtual_host: Virtual host name
:raises ApiError: Raises if the remote server encountered an error.
:raises ApiConnectionError: Raises if there was a connectivity issue.
:rtype: dict
"""
virtual_host = quote(virtual_host, '')
return self.http_client.get(API_USER_VIRTUAL_HOST_PERMISSIONS %
(
virtual_host,
username
))
def get_permissions(self, username):
"""Get all Users permissions.
:param str username: Username
:raises ApiError: Raises if the remote server encountered an error.
:raises ApiConnectionError: Raises if there was a connectivity issue.
:rtype: dict
"""
return self.http_client.get(API_USER_PERMISSIONS %
(
username
))
def set_permission(self, username, virtual_host, configure_regex='.*',
write_regex='.*', read_regex='.*'):
"""Set User permissions for the configured virtual host.
:param str username: Username
:param str virtual_host: Virtual host name
:param str configure_regex: Permission pattern for configuration
operations for this user.
:param str write_regex: Permission pattern for write operations
for this user.
:param str read_regex: Permission pattern for read operations
for this user.
:raises ApiError: Raises if the remote server encountered an error.
:raises ApiConnectionError: Raises if there was a connectivity issue.
:rtype: dict
"""
virtual_host = quote(virtual_host, '')
permission_payload = json.dumps({
"configure": configure_regex,
"read": read_regex,
"write": write_regex
})
return self.http_client.put(API_USER_VIRTUAL_HOST_PERMISSIONS %
(
virtual_host,
username
),
payload=permission_payload)
def delete_permission(self, username, virtual_host):
"""Delete User permissions for the configured virtual host.
:param str username: Username
:param str virtual_host: Virtual host name
:raises ApiError: Raises if the remote server encountered an error.
:raises ApiConnectionError: Raises if there was a connectivity issue.
:rtype: dict
"""
virtual_host = quote(virtual_host, '')
return self.http_client.delete(
API_USER_VIRTUAL_HOST_PERMISSIONS %
(
virtual_host,
username
))