156 lines
3.9 KiB
Python
156 lines
3.9 KiB
Python
"""
|
|
User Class object
|
|
"""
|
|
|
|
import logging
|
|
import requests
|
|
|
|
|
|
class User():
|
|
"""User class"""
|
|
|
|
def __init__(self, base_url, jwt):
|
|
"""
|
|
__init__ Init User
|
|
|
|
:param base_url: Base URL for User API
|
|
:param jwt: JWT Authentication for User API
|
|
"""
|
|
|
|
self.base_url = base_url
|
|
self.jwt = jwt
|
|
|
|
|
|
def get_all(self, args = None):
|
|
"""
|
|
get_all Get the users
|
|
|
|
:param args: Dict of args to add as URL params
|
|
"""
|
|
|
|
# Set the headers for the request
|
|
headers = {
|
|
'Content-Type': 'application/json',
|
|
'Accept': 'application/json',
|
|
'Authorization': f'Bearer {self.jwt}'
|
|
}
|
|
|
|
if not isinstance(args, dict):
|
|
args = {}
|
|
|
|
# Run the GET
|
|
response = requests.get(
|
|
f'{self.base_url}/api/v1/users',
|
|
params=args,
|
|
headers=headers,
|
|
timeout=300
|
|
)
|
|
|
|
# Good response. Return Events
|
|
if response.ok:
|
|
return response.json()
|
|
|
|
# Log error and return null
|
|
logging.error("[AFUSER GET ALL] [%s]", response.text)
|
|
return None
|
|
|
|
|
|
def get(self, user_key):
|
|
"""
|
|
get Get the user
|
|
|
|
:param user_key: The User key to get with API
|
|
"""
|
|
|
|
# Set the headers for the request
|
|
headers = {
|
|
'Content-Type': 'application/json',
|
|
'Accept': 'application/json',
|
|
'Authorization': f'Bearer {self.jwt}'
|
|
}
|
|
|
|
# Run the GET
|
|
response = requests.get(
|
|
f'{self.base_url}/api/v1/users/{user_key}',
|
|
headers=headers,
|
|
timeout=300
|
|
)
|
|
|
|
# Good response. Return User
|
|
if response.ok:
|
|
return response.json()
|
|
|
|
# Log error and return null
|
|
logging.error("[AFUSER GET] [%s]", response.text)
|
|
return None
|
|
|
|
|
|
def patch(self, user_key, patch_data):
|
|
"""
|
|
patch Patch the User
|
|
|
|
:param user_key: The User key to update with API
|
|
:param patch_data: The data to update the User
|
|
"""
|
|
|
|
# Set the headers for the request
|
|
headers = {
|
|
'Content-Type': 'application/json',
|
|
'Accept': 'application/json',
|
|
'Authorization': f'Bearer {self.jwt}'
|
|
}
|
|
|
|
# Set the key to the patch data
|
|
patch_data['key'] = user_key
|
|
|
|
# Run the GET
|
|
response = requests.patch(
|
|
f'{self.base_url}/api/v1/users/{user_key}',
|
|
headers=headers,
|
|
timeout=300,
|
|
json=patch_data
|
|
)
|
|
|
|
# Good response. Return User
|
|
if response.ok:
|
|
return response.json()
|
|
|
|
# Log error and return null
|
|
logging.error("[AFUSER UPDATE] [%s]", response.text)
|
|
return None
|
|
|
|
|
|
def change_role(self, user_key, role_type, role_key, action):
|
|
"""
|
|
patch Patch the User Roles
|
|
|
|
:param user_key: The User key to update with API
|
|
:param key_type: The type of key provided (either campaign, business, or company)
|
|
:param role_key: The Campaign / Business key to add to roles
|
|
:param action: The action to take (either add or remove)
|
|
"""
|
|
|
|
# verify parameters are not null
|
|
if user_key is None or role_key is None:
|
|
return None
|
|
if role_type not in ('campaign', 'business', 'company'):
|
|
return None
|
|
if action not in ('add', 'remove'):
|
|
return None
|
|
|
|
new_role = f"{role_type}_key:{role_key}"
|
|
|
|
# get the user record
|
|
user = self.get(user_key)
|
|
|
|
# check if role exists
|
|
if new_role in user['roles']:
|
|
if action == "remove":
|
|
user['roles'].remove(new_role)
|
|
else:
|
|
if action == "add":
|
|
user['roles'].append(new_role)
|
|
|
|
# patch the user record and return the results
|
|
return self.patch(user_key, {'roles': user['roles']})
|