Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@requires_key(KeenKeys.WRITE)
def post_events(self, events):
"""
Posts a single event to the Keen IO API. The write key must be set first.
:param events: an Event to upload
"""
url = "{0}/{1}/projects/{2}/events".format(self.base_url, self.api_version,
self.project_id)
headers = utilities.headers(self.write_key)
payload = json.dumps(events)
response = self.fulfill(HTTPMethods.POST, url, data=payload, headers=headers, timeout=self.post_timeout)
self._error_handling(response)
return self._get_response_json(response)
@requires_key(KeenKeys.READ)
def get_collection(self, event_collection):
"""
Extracts info about a collection using the Keen IO API. A master key must be set first.
:param event_collection: the name of the collection to retrieve info for
"""
url = "{0}/{1}/projects/{2}/events/{3}".format(self.base_url, self.api_version,
self.project_id, event_collection)
headers = utilities.headers(self.read_key)
response = self.fulfill(HTTPMethods.GET, url, headers=headers, timeout=self.get_timeout)
self._error_handling(response)
return response.json()
def method_wrapper(self, *args, **kwargs):
for case in switch(key_type):
if case(KeenKeys.READ):
if not self._get_read_key():
_throw_key_missing(KeenKeys.READ, bool(self._get_master_key()))
break
if case(KeenKeys.WRITE):
if not self._get_write_key():
_throw_key_missing(KeenKeys.WRITE, bool(self._get_master_key()))
break
if case(KeenKeys.MASTER):
if not self._get_master_key():
_throw_key_missing(KeenKeys.MASTER, False)
break
return func(self, *args, **kwargs)
@requires_key(KeenKeys.MASTER)
def update_access_key_options(self, access_key_id, options):
"""
Replaces all of the options on the access key but does not change
non-option properties such as permissions or the key's name.
:param access_key_id: the 'key' value of the access key to change the options of
:param options: the new dictionary of options for this key
"""
return self._update_access_key_pair(access_key_id, "options", options)
@requires_key(KeenKeys.MASTER)
def update(self, query_name, saved_query_attributes):
"""
Given a dict of attributes to be updated, update only those attributes
in the Saved Query at the resource given by 'query_name'. This will
perform two HTTP requests--one to fetch the query definition, and one
to set the new attributes. This method will intend to preserve any
other properties on the query.
Master key must be set.
"""
query_name_attr_name = "query_name"
refresh_rate_attr_name = "refresh_rate"
query_attr_name = "query"
metadata_attr_name = "metadata"