Refactor: create a Toggl API class (#27)

Author: thekaveman

compiler_admin/api/toggl.py

@@ -0,0 +1,112 @@
+from base64 import b64encode
+from datetime import datetime
+
+import requests
+
+from compiler_admin import __version__
+
+
+class Toggl:
+    """Toggl API Client.
+
+    See https://engineering.toggl.com/docs/.
+    """
+
+    API_BASE_URL = "https://api.track.toggl.com"
+    API_REPORTS_BASE_URL = "reports/api/v3"
+    API_WORKSPACE = "workspace/{}"
+    API_HEADERS = {"Content-Type": "application/json", "User-Agent": "compilerla/compiler-admin:{}".format(__version__)}
+
+    def __init__(self, api_token: str, workspace_id: int, **kwargs):
+        self._token = api_token
+        self.workspace_id = workspace_id
+
+        self.headers = dict(Toggl.API_HEADERS)
+        self.headers.update(self._authorization_header())
+
+        self.session = requests.Session()
+        self.session.headers.update(self.headers)
+        self.timeout = int(kwargs.get("timeout", 5))
+
+    @property
+    def workspace_url_fragment(self):
+        """The workspace portion of an API URL."""
+        return Toggl.API_WORKSPACE.format(self.workspace_id)
+
+    def _authorization_header(self):
+        """Gets an `Authorization: Basic xyz` header using the Toggl API token.
+
+        See https://engineering.toggl.com/docs/authentication.
+        """
+        creds = f"{self._token}:api_token"
+        creds64 = b64encode(bytes(creds, "utf-8")).decode("utf-8")
+        return {"Authorization": "Basic {}".format(creds64)}
+
+    def _make_report_url(self, endpoint: str):
+        """Get a fully formed URL for the Toggl Reports API v3 endpoint.
+
+        See https://engineering.toggl.com/docs/reports_start.
+        """
+        return "/".join((Toggl.API_BASE_URL, Toggl.API_REPORTS_BASE_URL, self.workspace_url_fragment, endpoint))
+
+    def detailed_time_entries(self, start_date: datetime, end_date: datetime, **kwargs) -> requests.Response:
+        """Request a CSV report from Toggl of detailed time entries for the given date range.
+
+        Args:
+            start_date (datetime): The beginning of the reporting period.
+
+            end_date (str): The end of the reporting period.
+
+        Extra `kwargs` are passed through as a POST json body.
+
+        By default, requests a report with the following configuration:
+            * `billable=True`
+            * `rounding=1` (True, but this is an int param)
+            * `rounding_minutes=15`
+
+        See https://engineering.toggl.com/docs/reports/detailed_reports#post-export-detailed-report.
+
+        Returns:
+            response (requests.Response): The HTTP response.
+        """
+        # ensure start_date precedes end_date
+        start_date, end_date = min(start_date, end_date), max(start_date, end_date)
+        start = start_date.strftime("%Y-%m-%d")
+        end = end_date.strftime("%Y-%m-%d")
+
+        # calculate a timeout based on the size of the reporting period in days
+        # approximately 5 seconds per month of query size, with a minimum of 5 seconds
+        range_days = (end_date - start_date).days
+        current_timeout = self.timeout
+        dynamic_timeout = int((max(30, range_days) / 30.0) * 5)
+        self.timeout = max(current_timeout, dynamic_timeout)
+
+        params = dict(
+            billable=True,
+            start_date=start,
+            end_date=end,
+            rounding=1,
+            rounding_minutes=15,
+        )
+        params.update(kwargs)
+
+        response = self.post_reports("search/time_entries.csv", **params)
+        self.timeout = current_timeout
+
+        return response
+
+    def post_reports(self, endpoint: str, **kwargs) -> requests.Response:
+        """Send a POST request to the Reports v3 `endpoint`.
+
+        Extra `kwargs` are passed through as a POST json body.
+
+        Will raise for non-200 status codes.
+
+        See https://engineering.toggl.com/docs/reports_start.
+        """
+        url = self._make_report_url(endpoint)
+
+        response = self.session.post(url, json=kwargs, timeout=self.timeout)
+        response.raise_for_status()
+
+        return response

compiler_admin/services/files.py

@@ -1,4 +1,6 @@
 import json
+import os
+from pathlib import Path
 
 import pandas as pd
 
@@ -23,3 +25,26 @@ def write_json(file_path: str, data):
     """Write a python object as JSON to the given path."""
     with open(file_path, "w") as f:
         json.dump(data, f, indent=2)
+
+
+class JsonFileCache:
+    """Very basic in-memory cache of a JSON file."""
+
+    def __init__(self, env_file_path=None):
+        self._cache = {}
+        self._path = None
+
+        if env_file_path:
+            p = os.environ.get(env_file_path)
+            self._path = Path(p) if p else None
+        if self._path and self._path.exists():
+            self._cache.update(read_json(self._path))
+
+    def __getitem__(self, key):
+        return self._cache.get(key)
+
+    def __setitem__(self, key, value):
+        self._cache[key] = value
+
+    def get(self, key, default=None):
+        return self._cache.get(key, default)

compiler_admin/services/toggl.py

@@ -1,27 +1,17 @@
-from base64 import b64encode
 from datetime import datetime
 import io
 import os
 import sys
 from typing import TextIO
 
 import pandas as pd
-import requests
 
-from compiler_admin import __version__
+from compiler_admin.api.toggl import Toggl
 from compiler_admin.services.google import user_info as google_user_info
 import compiler_admin.services.files as files
 
-# Toggl API config
-API_BASE_URL = "https://api.track.toggl.com"
-API_REPORTS_BASE_URL = "reports/api/v3"
-API_WORKSPACE = "workspace/{}"
-
-# cache of previously seen project information, keyed on Toggl project name
-PROJECT_INFO = {}
-
 # cache of previously seen user information, keyed on email
-USER_INFO = {}
+USER_INFO = files.JsonFileCache("TOGGL_USER_INFO")
 NOT_FOUND = "NOT FOUND"
 
 # input CSV columns needed for conversion
@@ -31,83 +21,9 @@
 OUTPUT_COLUMNS = ["Date", "Client", "Project", "Task", "Notes", "Hours", "First name", "Last name"]
 
 
-def _harvest_client_name():
-    """Gets the value of the HARVEST_CLIENT_NAME env var."""
-    return os.environ.get("HARVEST_CLIENT_NAME")
-
-
-def _get_info(obj: dict, key: str, env_key: str):
-    """Read key from obj, populating obj once from a file path at env_key."""
-    if obj == {}:
-        file_path = os.environ.get(env_key)
-        if file_path:
-            file_info = files.read_json(file_path)
-            obj.update(file_info)
-    return obj.get(key)
-
-
-def _toggl_api_authorization_header():
-    """Gets an `Authorization: Basic xyz` header using the Toggl API token.
-
-    See https://engineering.toggl.com/docs/authentication.
-    """
-    token = _toggl_api_token()
-    creds = f"{token}:api_token"
-    creds64 = b64encode(bytes(creds, "utf-8")).decode("utf-8")
-    return {"Authorization": "Basic {}".format(creds64)}
-
-
-def _toggl_api_headers():
-    """Gets a dict of headers for Toggl API requests.
-
-    See https://engineering.toggl.com/docs/.
-    """
-    headers = {"Content-Type": "application/json"}
-    headers.update({"User-Agent": "compilerla/compiler-admin:{}".format(__version__)})
-    headers.update(_toggl_api_authorization_header())
-    return headers
-
-
-def _toggl_api_report_url(endpoint: str):
-    """Get a fully formed URL for the Toggl Reports API v3 endpoint.
-
-    See https://engineering.toggl.com/docs/reports_start.
-    """
-    workspace_id = _toggl_workspace()
-    return "/".join((API_BASE_URL, API_REPORTS_BASE_URL, API_WORKSPACE.format(workspace_id), endpoint))
-
-
-def _toggl_api_token():
-    """Gets the value of the TOGGL_API_TOKEN env var."""
-    return os.environ.get("TOGGL_API_TOKEN")
-
-
-def _toggl_client_id():
-    """Gets the value of the TOGGL_CLIENT_ID env var."""
-    client_id = os.environ.get("TOGGL_CLIENT_ID")
-    if client_id:
-        return int(client_id)
-    return None
-
-
-def _toggl_project_info(project: str):
-    """Return the cached project for the given project key."""
-    return _get_info(PROJECT_INFO, project, "TOGGL_PROJECT_INFO")
-
-
-def _toggl_user_info(email: str):
-    """Return the cached user for the given email."""
-    return _get_info(USER_INFO, email, "TOGGL_USER_INFO")
-
-
-def _toggl_workspace():
-    """Gets the value of the TOGGL_WORKSPACE_ID env var."""
-    return os.environ.get("TOGGL_WORKSPACE_ID")
-
-
 def _get_first_name(email: str) -> str:
     """Get cached first name or derive from email."""
-    user = _toggl_user_info(email)
+    user = USER_INFO.get(email)
     first_name = user.get("First Name") if user else None
     if first_name is None:
         parts = email.split("@")
@@ -122,7 +38,7 @@ def _get_first_name(email: str) -> str:
 
 def _get_last_name(email: str):
     """Get cached last name or query from Google."""
-    user = _toggl_user_info(email)
+    user = USER_INFO.get(email)
     last_name = user.get("Last Name") if user else None
     if last_name is None:
         user = google_user_info(email)
@@ -134,7 +50,7 @@ def _get_last_name(email: str):
     return last_name
 
 
-def _str_timedelta(td):
+def _str_timedelta(td: str):
     """Convert a string formatted duration (e.g. 01:30) to a timedelta."""
     return pd.to_timedelta(pd.to_datetime(td, format="%H:%M:%S").strftime("%H:%M:%S"))
 
@@ -160,7 +76,7 @@ def convert_to_harvest(
         None. Either prints the resulting CSV data or writes to output_path.
     """
     if client_name is None:
-        client_name = _harvest_client_name()
+        client_name = os.environ.get("HARVEST_CLIENT_NAME")
 
     # read CSV file, parsing dates and times
     source = files.read_csv(source_path, usecols=INPUT_COLUMNS, parse_dates=["Start date"], cache_dates=True)
@@ -175,8 +91,9 @@ def convert_to_harvest(
     source["Client"] = client_name
     source["Task"] = "Project Consulting"
 
-    # get cached project name if any
-    source["Project"] = source["Project"].apply(lambda x: _toggl_project_info(x) or x)
+    # get cached project name if any, keyed on Toggl project name
+    project_info = files.JsonFileCache("TOGGL_PROJECT_INFO")
+    source["Project"] = source["Project"].apply(lambda x: project_info.get(key=x, default=x))
 
     # assign First and Last name
     source["First name"] = source["Email"].apply(_get_first_name)
@@ -208,42 +125,20 @@ def download_time_entries(
 
     Extra kwargs are passed along in the POST request body.
 
-    By default, requests a report with the following configuration:
-        * `billable=True`
-        * `client_ids=[$TOGGL_CLIENT_ID]`
-        * `rounding=1` (True, but this is an int param)
-        * `rounding_minutes=15`
-
-    See https://engineering.toggl.com/docs/reports/detailed_reports#post-export-detailed-report.
-
     Returns:
         None. Either prints the resulting CSV data or writes to output_path.
     """
-    start = start_date.strftime("%Y-%m-%d")
-    end = end_date.strftime("%Y-%m-%d")
-    # calculate a timeout based on the size of the reporting period in days
-    # approximately 5 seconds per month of query size, with a minimum of 5 seconds
-    range_days = (end_date - start_date).days
-    timeout = int((max(30, range_days) / 30.0) * 5)
-
-    if ("client_ids" not in kwargs or not kwargs["client_ids"]) and isinstance(_toggl_client_id(), int):
-        kwargs["client_ids"] = [_toggl_client_id()]
-
-    params = dict(
-        billable=True,
-        start_date=start,
-        end_date=end,
-        rounding=1,
-        rounding_minutes=15,
-    )
-    params.update(kwargs)
-
-    headers = _toggl_api_headers()
-    url = _toggl_api_report_url("search/time_entries.csv")
-
-    response = requests.post(url, json=params, headers=headers, timeout=timeout)
-    response.raise_for_status()
+    env_client_id = os.environ.get("TOGGL_CLIENT_ID")
+    if env_client_id:
+        env_client_id = int(env_client_id)
+    if ("client_ids" not in kwargs or not kwargs["client_ids"]) and isinstance(env_client_id, int):
+        kwargs["client_ids"] = [env_client_id]
+
+    token = os.environ.get("TOGGL_API_TOKEN")
+    workspace = os.environ.get("TOGGL_WORKSPACE_ID")
+    toggl = Toggl(token, workspace)
 
+    response = toggl.detailed_time_entries(start_date, end_date, **kwargs)
     # the raw response has these initial 3 bytes:
     #
     #   b"\xef\xbb\xbfUser,Email,Client..."