Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def delete(self, query_name):
"""
Deletes a saved query from a project with a query name.
Master key must be set.
"""
url = "{0}/{1}".format(self.saved_query_url, query_name)
response = self._get_json(HTTPMethods.DELETE, url, self._get_master_key())
return True
def delete_events(self, event_collection, params):
"""
Deletes events via the Keen IO API. A master key must be set first.
:param event_collection: string, the event collection from which event are being deleted
"""
url = "{0}/{1}/projects/{2}/events/{3}".format(self.base_url,
self.api_version,
self.project_id,
event_collection)
headers = utilities.headers(self.master_key)
response = self.fulfill(HTTPMethods.DELETE, url, params=params, headers=headers, timeout=self.post_timeout)
self._error_handling(response)
return True
:param name: the new name desired for this access key
:param is_active: whether the key should become enabled (True) or revoked (False)
:param permitted: the new list of permissions desired for this access key
:param options: the new dictionary of options for this access key
"""
url = "{0}/{1}/projects/{2}/keys/{3}".format(self.base_url, self.api_version,
self.project_id, access_key_id)
headers = utilities.headers(self.master_key)
payload_dict = {
"name": name,
"is_active": is_active,
"permitted": permitted,
"options": options
}
payload = json.dumps(payload_dict)
response = self.fulfill(HTTPMethods.POST, url, data=payload, headers=headers, timeout=self.get_timeout)
self._error_handling(response)
return response.json()
def create(self, query_name, saved_query):
"""
Creates the saved query via a PUT request to Keen IO Saved Query endpoint.
Master key must be set.
"""
url = "{0}/{1}".format(self.saved_query_url, query_name)
payload = saved_query
# To support clients that may have already called dumps() to work around how this used to
# work, make sure it's not a str. Hopefully it's some sort of mapping. When we actually
# try to send the request, client code will get an InvalidJSONError if payload isn't
# a json-formatted string.
if not isinstance(payload, str):
payload = json.dumps(saved_query)
response = self._get_json(HTTPMethods.PUT, url, self._get_master_key(), data=payload)
return response
def get(self, query_name):
"""
Gets a single saved query for a project from the Keen IO API given a
query name.
Master key must be set.
"""
url = "{0}/{1}".format(self.saved_query_url, query_name)
response = self._get_json(HTTPMethods.GET, url, self._get_master_key())
return response
def get(self, dataset_name):
""" Fetch a single Cached Dataset for a Project. Read key must be set.
:param dataset_name: Name of Cached Dataset (not `display_name`)
"""
url = "{0}/{1}".format(self._cached_datasets_url, dataset_name)
return self._get_json(HTTPMethods.GET, url, self._get_read_key())
def query(self, analysis_type, params, all_keys=False):
"""
Performs a query using the Keen IO analysis API. A read key must be set first.
"""
if not self._order_by_is_valid_or_none(params):
raise ValueError("order_by given is invalid or is missing required group_by.")
if not self._limit_is_valid_or_none(params):
raise ValueError("limit given is invalid or is missing required order_by.")
url = "{0}/{1}/projects/{2}/queries/{3}".format(self.base_url, self.api_version,
self.project_id, analysis_type)
headers = utilities.headers(self.read_key)
payload = params
response = self.fulfill(HTTPMethods.GET, url, params=payload, headers=headers, timeout=self.get_timeout)
self._error_handling(response)
response = response.json()
if not all_keys:
response = response["result"]
return response