add support for from_ts and to_ts, add replication_key

tap_datadog/client.py

@@ -74,10 +74,10 @@ def get_url_params(
         if self.replication_key:
             params["sort"] = "asc"
             params["order_by"] = self.replication_key
-        to_ts = datetime.now()
-        from_ts = datetime.now() - timedelta(days=1)
-        params["from_ts"] = int(from_ts.timestamp())
-        params["to_ts"] = int(to_ts.timestamp())
+        to_ts = self.config.get("to_ts") if self.config.get("to_ts") != None else datetime.now().timestamp()
+        from_ts = self.config.get("from_ts") if self.config.get("from_ts") != None else (datetime.now() - timedelta(days=1)).timestamp()
+        params["from_ts"] = int(from_ts)
+        params["to_ts"] = int(to_ts)
         return params
 
     def parse_response(self, response: requests.Response) -> Iterable[dict]:

tap_datadog/streams.py

@@ -21,7 +21,8 @@ class SLOStream(DatadogStream):
 
     path = "slo/{id}/history"
     primary_keys: t.ClassVar[list[str]] = ["slo_id"]
-    replication_key = None
+    replication_key = "to_ts"
+    is_sorted = True
     # Optionally, you may also use `schema_filepath` in place of `schema`:
     # schema_filepath = SCHEMAS_DIR / "users.json"  # noqa: ERA001
     schema_filepath = SCHEMAS_DIR / "slo_history.json"

tap_datadog/tap.py

@@ -40,6 +40,27 @@ class Tapdatadog(Tap):
             required=True,
             description="SLOs to replicate",
         ),
+        th.Property(
+            "from_ts",
+            th.NumberType,
+            required=False,
+            description="The Unix timestamp to start the extract from, by default now - 1 day",
+        ),
+        th.Property(
+            "to_ts",
+            th.NumberType,
+            required=False,
+            description="The Unix timestamp to start the extract to, by default now",
+        ),
+        th.Property(
+            "slos",
+            th.ArrayType(th.ObjectType(
+                th.Property("name", th.StringType, description="The name of the SLO"),
+                th.Property("id", th.StringType, description="The id of the SLO"),
+            )),
+            required=True,
+            description="SLOs to replicate",
+        ),
     ).to_dict()
 
     def discover_streams(self) -> list[streams.datadogStream]: