168 lines
4.3 KiB
Python
168 lines
4.3 KiB
Python
"""
|
|
Event Class object
|
|
"""
|
|
|
|
import logging
|
|
import requests
|
|
|
|
|
|
class Event():
|
|
"""Event class"""
|
|
|
|
def __init__(self, base_url, jwt):
|
|
"""
|
|
__init__ Init Event
|
|
|
|
:param base_url: Base URL for Event API
|
|
:param jwt: JWT Authentication for Event API
|
|
"""
|
|
|
|
self.base_url = base_url
|
|
self.jwt = jwt
|
|
|
|
|
|
def get_all(self, args = None):
|
|
"""
|
|
get_all Get the events
|
|
|
|
: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/events',
|
|
params=args,
|
|
headers=headers,
|
|
timeout=300
|
|
)
|
|
|
|
# Good response. Return Events
|
|
if response.ok:
|
|
return response.json()
|
|
|
|
# Log error and retunr null
|
|
logging.error("[AFEVENT GET ALL] [%s]", response.text)
|
|
return None
|
|
|
|
|
|
def get(self, event_key):
|
|
"""
|
|
get Get the event
|
|
|
|
:param event_key: The Event 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/events/{event_key}',
|
|
headers=headers,
|
|
timeout=300
|
|
)
|
|
|
|
# Good response. Return Event
|
|
if response.ok:
|
|
return response.json()
|
|
|
|
# Log error and retunr null
|
|
logging.error("[AFEVENT GET] [%s]", response.text)
|
|
return None
|
|
|
|
|
|
def patch(self, event_key, patch_data):
|
|
"""
|
|
patch Path the Event
|
|
|
|
:param event_key: The Event key to update with API
|
|
:param patch_data: The data to update the Event
|
|
"""
|
|
|
|
# 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'] = event_key
|
|
|
|
# Run the GET
|
|
response = requests.patch(
|
|
f'{self.base_url}/api/v1/events/{event_key}',
|
|
headers=headers,
|
|
timeout=300,
|
|
json=patch_data
|
|
)
|
|
|
|
# Good response. Return Event
|
|
if response.ok:
|
|
return response.json()
|
|
|
|
# Log error and retunr null
|
|
logging.error("[AFEVENT UPDATE] [%s]", response.text)
|
|
try:
|
|
return response.json()
|
|
except: # pylint: disable=bare-except
|
|
return None
|
|
|
|
|
|
def patch_v2(self, event_key, patch_data, retry=-1):
|
|
"""
|
|
patch_v2 Path the Event V2. This will retry a 409 as many times as set by 'retry' parameter
|
|
|
|
:param event_key: The Event key to update with API
|
|
:param patch_data: The data to update the Event
|
|
:param retry: If set to a positive number an attempt to replay the path will happen
|
|
"""
|
|
|
|
# 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'] = event_key
|
|
|
|
# Run the GET
|
|
response = requests.patch(
|
|
f'{self.base_url}/api/v1/events/{event_key}',
|
|
headers=headers,
|
|
timeout=30,
|
|
json=patch_data
|
|
)
|
|
|
|
# Good response. Return Event
|
|
if response.ok:
|
|
return response.json(), response.status_code
|
|
|
|
# Log error and retunr null
|
|
logging.error("[AFEVENT UPDATE] [%s]", response.text)
|
|
|
|
# Check if this call should be tried again
|
|
if str(response.status_code) == "409" and retry > 0:
|
|
return self.patch_v2(event_key, patch_data, retry=retry-1)
|
|
|
|
try:
|
|
return response.json(), response.status_code
|
|
except: # pylint: disable=bare-except
|
|
return None, response.status_code
|