Merge pull request #2 from harrystech/ynaim94/deng-2040/add-topic-stream

DENG-2040: add export topics.jsonl stream

pyproject.toml

@@ -13,18 +13,20 @@ license = "Apache 2.0"
 python = "<3.11,>=3.7.1"
 requests = "^2.25.1"
 singer-sdk = "^0.10.0"
-black = "^22.8.0"
-importlib-metadata = "^4.12.0"
+pendulum = "^2.1.2"
 
 [tool.poetry.dev-dependencies]
 pytest = "^6.2.5"
 tox = "^3.24.4"
 flake8 = "^3.9.2"
-black = "^22.3.0"
 pydocstyle = "^6.1.1"
 mypy = "^0.910"
 types-requests = "^2.26.1"
 isort = "^5.10.1"
+importlib-metadata = "^4.12.0"
+black = "^22.8.0"
+
+
 
 [tool.isort]
 profile = "black"

tap_gladly/client.py

@@ -13,8 +13,6 @@
 class gladlyStream(RESTStream):
     """gladly stream class."""
 
-    _common_date_format = "%Y-%m-%dT%H:%M:%SZ"
-
     # OR use a dynamic url_base:
     @property
     def url_base(self) -> str:
@@ -75,8 +73,3 @@ def parse_response(self, response: requests.Response) -> Iterable[dict]:
         """Parse the response and return an iterator of result records."""
         # TODO: Parse response body and return a set of records.
         yield from extract_jsonpath(self.records_jsonpath, input=response.json())
-
-    def post_process(self, row, context):
-        """As needed, append or transform raw data to match expected structure."""
-        # TODO: Delete this method if not needed.
-        return row

tap_gladly/schemas/export_topics.json

@@ -0,0 +1,18 @@
+{
+  "$schema": "http://json-schema.org/schema#",
+  "type": "object",
+  "properties": {
+    "id": {
+      "type": "string"
+    },
+    "disabled": {
+      "type": "boolean"
+    },
+    "name": {
+      "type": "string"
+    },
+    "parentId": {
+      "type": "string"
+    }
+  }
+}

tap_gladly/streams.py

@@ -1,38 +1,34 @@
 """Stream type classes for tap-gladly."""
 import abc
 import json
-from datetime import datetime
 from pathlib import Path
 from typing import Iterable, Optional
 
+import pendulum
 import requests
 from singer_sdk import exceptions
 from singer_sdk.helpers.jsonpath import extract_jsonpath
 
 from tap_gladly.client import gladlyStream
 
-# TODO: Delete this is if not using json files for schema definition
 SCHEMAS_DIR = Path(__file__).parent / Path("./schemas")
 
 
-class ExportJobsStream(gladlyStream):
+class ExportCompletedJobsStream(gladlyStream):
     """List export jobs stream."""
 
     name = "jobs"
-    path = "/export/jobs"
+    path = "/export/jobs?status=COMPLETED"
     primary_keys = ["id"]
     replication_key = None
     # Optionally, you may also use `schema_filepath` in place of `schema`:
     schema_filepath = SCHEMAS_DIR / "export_jobs.json"
 
-    # start_date
     def post_process(self, row, context):
-        """As needed, append or transform raw data to match expected structure."""
-        if "start_date" not in self.config:
-            return row
-        if datetime.strptime(
-            row["parameters"]["startAt"], self._common_date_format
-        ) > datetime.strptime(self.config["start_date"], self._common_date_format):
+        """Filter jobs that finished before start_date."""
+        if pendulum.parse(row["parameters"]["endAt"]) >= pendulum.parse(
+            self.config["start_date"]
+        ):
             return row
         return
 
@@ -41,14 +37,31 @@ def get_child_context(self, record: dict, context: Optional[dict]) -> dict:
         return {"job_id": record["id"]}
 
 
+class ExportFileTopicsStream(gladlyStream):
+    """Abstract class, export conversation items stream."""
+
+    name = "topics"
+    path = "/export/jobs/{job_id}/files/topics.jsonl"
+    primary_keys = ["id"]
+    replication_key = None
+    parent_stream_type = ExportCompletedJobsStream
+    ignore_parent_replication_key = True
+    schema_filepath = SCHEMAS_DIR / "export_topics.json"
+
+    def parse_response(self, response: requests.Response) -> Iterable[dict]:
+        """Parse the response and return an iterator of result records."""
+        for line in response.iter_lines():
+            yield from extract_jsonpath(self.records_jsonpath, input=json.loads(line))
+
+
 class ExportFileConversationItemsStream(gladlyStream, abc.ABC):
     """Abstract class, export conversation items stream."""
 
     name = "conversation_conversation_items"
     path = "/export/jobs/{job_id}/files/conversation_items.jsonl"
     primary_keys = ["id"]
     replication_key = None
-    parent_stream_type = ExportJobsStream
+    parent_stream_type = ExportCompletedJobsStream
     ignore_parent_replication_key = True
 
     @property

tap_gladly/tap.py

@@ -7,25 +7,27 @@
 
 # TODO: Import your custom stream types here:
 from tap_gladly.streams import (
+    ExportCompletedJobsStream,
     ExportFileConversationItemsChatMessage,
     ExportFileConversationItemsConversationNote,
     ExportFileConversationItemsConversationStatusChange,
     ExportFileConversationItemsPhoneCall,
     ExportFileConversationItemsSms,
     ExportFileConversationItemsTopicChange,
     ExportFileConversationItemsVoiceMail,
-    ExportJobsStream,
+    ExportFileTopicsStream,
 )
 
 STREAM_TYPES = [
-    ExportJobsStream,
+    ExportCompletedJobsStream,
     ExportFileConversationItemsChatMessage,
     ExportFileConversationItemsConversationNote,
     ExportFileConversationItemsTopicChange,
     ExportFileConversationItemsSms,
     ExportFileConversationItemsConversationStatusChange,
     ExportFileConversationItemsPhoneCall,
     ExportFileConversationItemsVoiceMail,
+    ExportFileTopicsStream,
 ]
 
 

tap_gladly/tests/test_core.py

@@ -1,16 +1,13 @@
 """Tests standard tap features using the built-in SDK tests library."""
 
-import datetime
 
+import pendulum
 from singer_sdk.testing import get_standard_tap_tests
 
-from tap_gladly.client import gladlyStream
 from tap_gladly.tap import Tapgladly
 
 SAMPLE_CONFIG = {
-    "start_date": datetime.datetime.now(datetime.timezone.utc).strftime(
-        gladlyStream._common_date_format
-    ),
+    "start_date": pendulum.now().isoformat(),
     "username": "test",
     "password": "test",
     "api_url_base": "api_base_url",