Merge pull request #12 from Widen/fix-time-entries

flattened records in the time_entries stream

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "acquia-tap-toggl"
-version = "0.0.1"
+version = "0.1.0"
 description = "`tap-toggl` is a Singer tap for Toggl, built with the Meltano Singer SDK."
 readme = "README.md"
 authors = ["Josh Lloyd <[email protected]>"]

tap_toggl/client.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import sys
 from base64 import b64encode
 from datetime import datetime
 from typing import Any, Callable
@@ -17,13 +16,13 @@
 class TogglStream(RESTStream):
     """Toggl stream class."""
 
+    records_jsonpath = "$[*]"
+
     @property
     def url_base(self) -> str:
         """Return the API URL root, configurable via tap settings."""
         return "https://api.track.toggl.com"
 
-    records_jsonpath = "$[*]"
-
     @property
     def http_headers(self) -> dict:
         """Return the http headers needed.

tap_toggl/streams.py

@@ -5,7 +5,9 @@
 import sys
 import typing as t
 
+import requests
 from singer_sdk import typing as th
+from singer_sdk.helpers.jsonpath import extract_jsonpath
 from singer_sdk.pagination import BaseAPIPaginator, SimpleHeaderPaginator
 
 from tap_toggl.client import TogglStream, TogglPaginationStream
@@ -338,7 +340,7 @@ class TimeEntriesStream(TogglStream):
     name = "time_entries"
     path = "/reports/api/v3/workspace/{workspace_id}/search/time_entries"
     rest_method = "POST"
-    primary_keys: t.ClassVar[list[str]] = ["time_entries"]
+    primary_keys: t.ClassVar[list[str]] = ["id"]
     replication_key = None
     schema = th.PropertiesList(
         th.Property("billable", th.BooleanType),
@@ -350,18 +352,11 @@ class TimeEntriesStream(TogglStream):
         th.Property("row_number", th.IntegerType),
         th.Property("tag_ids", th.ArrayType(th.IntegerType)),
         th.Property("task_id", th.IntegerType),
-        th.Property(
-            "time_entries",
-            th.ArrayType(
-                th.ObjectType(
-                    th.Property("at", th.DateTimeType),
-                    th.Property("id", th.IntegerType),
-                    th.Property("seconds", th.IntegerType),
-                    th.Property("start", th.DateTimeType),
-                    th.Property("stop", th.DateTimeType),
-                )
-            ),
-        ),
+        th.Property("at", th.DateTimeType),
+        th.Property("id", th.IntegerType),
+        th.Property("seconds", th.IntegerType),
+        th.Property("start", th.DateTimeType),
+        th.Property("stop", th.DateTimeType),
         th.Property("user_id", th.IntegerType),
         th.Property("username", th.StringType),
         th.Property("workspace_id", th.IntegerType),
@@ -397,6 +392,23 @@ def get_new_paginator(self) -> BaseAPIPaginator:
         """
         return SimpleHeaderPaginator("X-Next-Row-Number")
 
+    def parse_response(self, response: requests.Response) -> t.Iterable[dict]:
+        """Parse the response and return an iterator of result records.
+
+        Args:
+            response: A raw :class:`requests.Response`
+
+        Yields:
+            One item for every item found in the response.
+        """
+        data = response.json()
+        records = extract_jsonpath(self.records_jsonpath, input=data)
+        for record in records:
+            for time_entry in record.get('time_entries', []):
+                del record["time_entries"]
+                record = {**record, **time_entry}
+                yield record
+
     def post_process(
         self,
         row: dict,

tests/test_core.py

@@ -13,13 +13,11 @@
 }
 
 TEST_SUITE_CONFIG = SuiteConfig(
-    ignore_no_records_for_streams=["projects", "tasks", "time_entries"]
+    ignore_no_records_for_streams=["projects", "tasks", "time_entries", "tags"]
 )
 
 TestTapToggl = get_tap_test_class(
     tap_class=TapToggl,
     config=SAMPLE_CONFIG,
     suite_config=TEST_SUITE_CONFIG,
 )
-
-# TODO: Create additional tests as appropriate for your tap.