Only extract from completed jobs

harrystech · Sep 23, 2022 · 463d7df · 463d7df
1 parent 7701b37
commit 463d7df
Show file tree

Hide file tree

Showing 3 changed files with 15 additions and 15 deletions.
diff --git a/tap_gladly/streams.py b/tap_gladly/streams.py
@@ -14,22 +14,19 @@
 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"
 
     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 pendulum.parse(row["parameters"]["startAt"]) >= pendulum.parse(
+        """Filter jobs that finished before start_date."""
+        if pendulum.parse(row["parameters"]["endAt"]) >= pendulum.parse(
             self.config["start_date"]
         ):
             return row
@@ -47,7 +44,7 @@ class ExportFileTopicsStream(gladlyStream):
     path = "/export/jobs/{job_id}/files/topics.jsonl"
     primary_keys = ["id"]
     replication_key = None
-    parent_stream_type = ExportJobsStream
+    parent_stream_type = ExportCompletedJobsStream
     ignore_parent_replication_key = True
     schema_filepath = SCHEMAS_DIR / "export_topics.json"
 
@@ -64,7 +61,7 @@ class ExportFileConversationItemsStream(gladlyStream, abc.ABC):
     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

diff --git a/tap_gladly/tap.py b/tap_gladly/tap.py
@@ -7,6 +7,7 @@
 
 # TODO: Import your custom stream types here:
 from tap_gladly.streams import (
+    ExportCompletedJobsStream,
     ExportFileConversationItemsChatMessage,
     ExportFileConversationItemsConversationNote,
     ExportFileConversationItemsConversationStatusChange,
@@ -15,11 +16,10 @@
     ExportFileConversationItemsTopicChange,
     ExportFileConversationItemsVoiceMail,
     ExportFileTopicsStream,
-    ExportJobsStream,
 )
 
 STREAM_TYPES = [
-    ExportJobsStream,
+    ExportCompletedJobsStream,
     ExportFileConversationItemsChatMessage,
     ExportFileConversationItemsConversationNote,
     ExportFileConversationItemsTopicChange,

diff --git a/tap_gladly/tests/test_streams.py b/tap_gladly/tests/test_streams.py
@@ -4,7 +4,10 @@
 
 import pendulum
 
-from tap_gladly.streams import ExportFileConversationItemsChatMessage, ExportJobsStream
+from tap_gladly.streams import (
+    ExportCompletedJobsStream,
+    ExportFileConversationItemsChatMessage,
+)
 from tap_gladly.tap import Tapgladly
 
 SAMPLE_CONFIG = {
@@ -23,16 +26,16 @@ def test_started_at():
         ),
         parse_env_config=False,
     )
-    export_jobs_stream = ExportJobsStream(tap_gladly)
+    export_jobs_stream = ExportCompletedJobsStream(tap_gladly)
     before_row = {
         "record": "data",
         "parameters": {
-            "startAt": (pendulum.now() - datetime.timedelta(days=3)).isoformat()
+            "endAt": (pendulum.now() - datetime.timedelta(days=3)).isoformat()
         },
     }
     after_row = {
         "record": "data",
-        "parameters": {"startAt": pendulum.now().isoformat()},
+        "parameters": {"endAt": pendulum.now().isoformat()},
     }
     assert not export_jobs_stream.post_process(before_row, None)
     assert export_jobs_stream.post_process(after_row, None)