Created
February 26, 2016 17:34
-
-
Save seufagner/2eb7c53d291d637733e0 to your computer and use it in GitHub Desktop.
Wrapper to consume data from Localytics REST API
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| # The API enforces the following rate limits per API key on usage: | |
| # | |
| # Allows one simultaneous request | |
| # 60 requests/minute | |
| # 150 requests/hour | |
| # 1000 requests/day | |
| # The API will return a 429 response code when a rate limit is exceeded. | |
| # You should reduce your rate of requests and retry throttled requests after the | |
| # specified time. | |
| # The Localytics API, and the backend services which power it, are organized around | |
| # metrics, dimensions, and conditions. | |
| # In a nutshell, rows are selected according to conditions, | |
| # then grouped by dimensions, and finally tallied by metric. | |
| import os | |
| import requests | |
| from requests.auth import HTTPBasicAuth | |
| import json | |
| LOCALYTICS_API_KEY = os.getenv("LOCALYTICS_API_KEY") | |
| LOCALYTICS_API_SECRET = os.getenv("LOCALYTICS_API_SECRET") | |
| LOCALYTICS_DEFAULT_APP_ID = os.getenv("LOCALYTICS_DEFAULT_APP_ID") | |
| QUERY_ROOT_URL = "https://api.localytics.com/v1/query" | |
| class Localytics(object): | |
| def __init__(self, root_url=QUERY_ROOT_URL, key=LOCALYTICS_API_KEY, secret=LOCALYTICS_API_SECRET): | |
| self._root_url = root_url | |
| self._key = key | |
| self._secret = secret | |
| self._metrics = None | |
| self._dimensions = None | |
| self._conditions = None | |
| self._app_id = None | |
| def choose_app(self, app_id): | |
| self._app_id = app_id | |
| return self | |
| def metrics_from(self, metrics): | |
| """ | |
| :param metrics: could be single string or string array | |
| :return: itself | |
| """ | |
| self._metrics = metrics | |
| return self | |
| def conditions(self, conditions, as_dict=True): | |
| """ | |
| :param conditions: clausules to query result | |
| :param as_dict: conditions comes as dict type. | |
| :return: itself | |
| """ | |
| self._conditions = json.dumps(conditions) if as_dict else conditions | |
| return self | |
| def grouped_by(self, dimensions): | |
| """ | |
| :param dimensions: attribute to group results | |
| :return: itself | |
| """ | |
| self._dimensions = dimensions | |
| return self | |
| def fetch_json(self): | |
| """ | |
| Make requests using HTTP GET | |
| :return: value to result key as json type. | |
| """ | |
| query = self._build_query() | |
| result = requests.get(self._root_url, | |
| auth=HTTPBasicAuth(self._key, self._secret), | |
| headers={'Accept':'application/vnd.localytics.v1+hal+json'}, | |
| params=query | |
| ) | |
| self._metrics = None | |
| self._dimensions = None | |
| self._conditions = None | |
| if result.status_code == 200: | |
| return json.loads(result.content)["results"] | |
| else: | |
| return None | |
| def _build_query(self): | |
| return { | |
| "app_id":self._app_id if self._app_id else LOCALYTICS_DEFAULT_APP_ID, | |
| "dimensions":self._dimensions, | |
| "metrics":self._metrics, | |
| "conditions":self._conditions | |
| } | |
| if __name__ == '__main__': | |
| localytics = Localytics() | |
| result = (localytics | |
| .choose_app(LOCALYTICS_DEFAULT_APP_ID) | |
| .metrics_from("users") | |
| .conditions({"event_name":["IN", "Search"], | |
| "day":["between","2016-02-22", "2016-02-25"]} | |
| ) | |
| .grouped_by(["a:searchType"]) | |
| ).fetch_json() | |
| print result |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment