refactor(toggl): WIP create an API class

Author: thekaveman

compiler_admin/services/toggl.py

@@ -12,11 +12,6 @@
 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 = {}
 
@@ -31,6 +26,108 @@
 OUTPUT_COLUMNS = ["Date", "Client", "Project", "Task", "Notes", "Hours", "First name", "Last name"]
 
 
+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(self.API_HEADERS)
+        self.headers.update(self._authorization_header())
+
+        self.timeout = int(kwargs.get("timeout", 5))
+
+    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((self.API_BASE_URL, self.API_REPORTS_BASE_URL, self.workspace_url_fragment, endpoint))
+
+    @property
+    def workspace_url_fragment(self):
+        """The workspace portion of an API URL."""
+        return self.API_WORKSPACE.format(self.workspace_id)
+
+    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 = requests.post(url, json=kwargs, headers=self.headers, timeout=self.timeout)
+        response.raise_for_status()
+
+        return response
+
+    def detailed_time_entries(self, start_date: datetime, end_date: datetime, **kwargs):
+        """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.
+        """
+        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 _harvest_client_name():
     """Gets the value of the HARVEST_CLIENT_NAME env var."""
     return os.environ.get("HARVEST_CLIENT_NAME")
@@ -46,37 +143,6 @@ def _get_info(obj: dict, key: str, env_key: str):
     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")
@@ -208,42 +274,17 @@ 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()
+    token = _toggl_api_token()
+    workspace = _toggl_workspace()
+    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..."

tests/services/test_toggl.py

@@ -7,6 +7,7 @@
 import pandas as pd
 import pytest
 
+from compiler_admin import __version__
 import compiler_admin.services.toggl
 from compiler_admin.services.toggl import (
     __name__ as MODULE,
@@ -15,6 +16,7 @@
     OUTPUT_COLUMNS,
     PROJECT_INFO,
     USER_INFO,
+    Toggl,
     _harvest_client_name,
     _get_info,
     _toggl_project_info,
@@ -60,28 +62,111 @@ def mock_google_user_info(mocker):
 
 
 @pytest.fixture
-def mock_api_env(monkeypatch):
+def mock_requests(mocker):
+    return mocker.patch(f"{MODULE}.requests")
+
+
+@pytest.fixture
+def mock_toggl_api(mocker):
+    Toggl = mocker.patch(f"{MODULE}.Toggl")
+    return Toggl.return_value
+
+
+@pytest.fixture
+def mock_toggl_api_env(monkeypatch):
     monkeypatch.setenv("TOGGL_API_TOKEN", "token")
     monkeypatch.setenv("TOGGL_CLIENT_ID", "1234")
     monkeypatch.setenv("TOGGL_WORKSPACE_ID", "workspace")
 
 
 @pytest.fixture
-def mock_requests(mocker):
-    return mocker.patch(f"{MODULE}.requests")
+def toggl():
+    return Toggl("token", 1234)
 
 
 @pytest.fixture
-def mock_api_post(mocker, mock_requests, toggl_file):
+def toggl_mock_post_reports(mocker, toggl, toggl_file):
     # setup a mock response to a requests.post call
     mock_csv_bytes = Path(toggl_file).read_bytes()
     mock_post_response = mocker.Mock()
     mock_post_response.raise_for_status.return_value = None
     # prepend the BOM to the mock content
     mock_post_response.content = b"\xef\xbb\xbf" + mock_csv_bytes
     # override the requests.post call to return the mock response
-    mock_requests.post.return_value = mock_post_response
-    return mock_requests
+    mocker.patch.object(toggl, "post_reports", return_value=mock_post_response)
+    return toggl
+
+
+@pytest.fixture
+def toggl_mock_detailed_time_entries(mock_toggl_api, toggl_file):
+    mock_csv_bytes = Path(toggl_file).read_bytes()
+    mock_toggl_api.detailed_time_entries.return_value.content = mock_csv_bytes
+    return mock_toggl_api
+
+
+def test_toggl_init(toggl):
+    token64 = "dG9rZW46YXBpX3Rva2Vu"
+
+    assert toggl._token == "token"
+    assert toggl.workspace_id == 1234
+    assert toggl.workspace_url_fragment == "workspace/1234"
+
+    assert toggl.headers["Content-Type"] == "application/json"
+
+    user_agent = toggl.headers["User-Agent"]
+    assert "compilerla/compiler-admin" in user_agent
+    assert __version__ in user_agent
+
+    assert toggl.headers["Authorization"] == f"Basic {token64}"
+
+    assert toggl.timeout == 5
+
+
+def test_toggl_make_report_url(toggl):
+    url = toggl._make_report_url("endpoint")
+
+    assert url.startswith(toggl.API_BASE_URL)
+    assert toggl.API_REPORTS_BASE_URL in url
+    assert toggl.workspace_url_fragment in url
+    assert "/endpoint" in url
+
+
+def test_toggl_post_reports(mock_requests, toggl):
+    url = toggl._make_report_url("endpoint")
+    response = toggl.post_reports("endpoint", kwarg1=1, kwarg2="two")
+
+    response.raise_for_status.assert_called_once()
+
+    mock_requests.post.assert_called_once_with(
+        url, json=dict(kwarg1=1, kwarg2="two"), headers=toggl.headers, timeout=toggl.timeout
+    )
+
+
+def test_toggl_detailed_time_entries(toggl_mock_post_reports):
+    dt = datetime(2024, 9, 25)
+    toggl_mock_post_reports.detailed_time_entries(dt, dt, kwarg1=1, kwarg2="two")
+
+    toggl_mock_post_reports.post_reports.assert_called_once_with(
+        "search/time_entries.csv",
+        billable=True,
+        start_date="2024-09-25",
+        end_date="2024-09-25",
+        rounding=1,
+        rounding_minutes=15,
+        kwarg1=1,
+        kwarg2="two",
+    )
+
+
+def test_toggl_detailed_time_entries_dynamic_timeout(mock_requests, toggl):
+    # range of 6 months
+    # timeout should be 6 * 5 = 30
+    start = datetime(2024, 1, 1)
+    end = datetime(2024, 6, 30)
+    toggl.detailed_time_entries(start, end)
+
+    mock_requests.post.assert_called_once()
+    assert mock_requests.post.call_args.kwargs["timeout"] == 30
 
 
 def test_harvest_client_name(monkeypatch):
@@ -241,27 +326,13 @@ def test_convert_to_harvest_sample(toggl_file, harvest_file, mock_google_user_in
     assert output_df["Client"].eq("Test Client 123").all()
 
 
-@pytest.mark.usefixtures("mock_api_env")
-def test_download_time_entries(toggl_file, mock_api_post):
+@pytest.mark.usefixtures("mock_toggl_api_env", "toggl_mock_detailed_time_entries")
+def test_download_time_entries(toggl_file):
     dt = datetime.now()
     mock_csv_bytes = Path(toggl_file).read_bytes()
 
     with NamedTemporaryFile("w") as temp:
-        download_time_entries(dt, dt, temp.name, extra_1=1, extra_2="two")
-
-        json_params = mock_api_post.post.call_args.kwargs["json"]
-        assert isinstance(json_params, dict)
-        assert json_params["billable"] is True
-        assert json_params["client_ids"] == [1234]
-        assert json_params["end_date"] == dt.strftime("%Y-%m-%d")
-        assert json_params["extra_1"] == 1
-        assert json_params["extra_2"] == "two"
-        assert json_params["rounding"] == 1
-        assert json_params["rounding_minutes"] == 15
-        assert json_params["start_date"] == dt.strftime("%Y-%m-%d")
-
-        assert mock_api_post.post.call_args.kwargs["timeout"] == 5
-
+        download_time_entries(dt, dt, temp.name)
         temp.flush()
         response_csv_bytes = Path(temp.name).read_bytes()
 
@@ -276,15 +347,3 @@ def test_download_time_entries(toggl_file, mock_api_post):
         # as corresponding column values from the mock DataFrame
         for col in response_df.columns:
             assert response_df[col].equals(mock_df[col])
-
-
-@pytest.mark.usefixtures("mock_api_env")
-def test_download_time_entries_dynamic_timeout(mock_api_post):
-    # range of 6 months
-    # timeout should be 6 * 5 = 30
-    start = datetime(2024, 1, 1)
-    end = datetime(2024, 6, 30)
-
-    download_time_entries(start, end)
-
-    assert mock_api_post.post.call_args.kwargs["timeout"] == 30