refactor: change from sync to bulk endpoints for transactions_lines and gl_accounts. (#294)

* refactor: change from sync to bulk endpoints for `transactions_lines` and `gl_accounts`.

all other streams will be full load. this will allow us to have more control over the start timestamp and it will now include all data, also deleted records.

* remove import

* account for none start date

* change strftime place

* fix f string

* fix if statement

* fix linting

* last linting

Author: tobiascadee

tap_exact/client.py

@@ -11,16 +11,13 @@
 import requests
 import xmltodict
 from lxml import etree
-from pendulum import parse
 from singer_sdk.helpers.jsonpath import extract_jsonpath
 from singer_sdk.pagination import BaseOffsetPaginator
 from singer_sdk.streams import RESTStream
 
 from tap_exact.auth import ExactAuthenticator
 
 if typing.TYPE_CHECKING:
-    from datetime import datetime
-
     from requests import Response
 
 TPageToken = typing.TypeVar("TPageToken")
@@ -100,22 +97,14 @@ def authenticator(self) -> _Auth:
         """
         return ExactAuthenticator(self)
 
-    def get_starting_time(self, context: dict) -> datetime:
-        """Return the starting timestamp for the stream."""
-        start_date = self.config.get("start_date")
-        if start_date:
-            start_date = parse(self.config.get("start_date"))
-        replication_key = self.get_starting_timestamp(context)
-        return replication_key or start_date
-
     def get_url_params(self, context: dict | None, next_page_token: str) -> dict[str, Any]:
         """Return a dictionary of parameters to use in the API request."""
         params: dict = {}
         if self.select:
             params["$select"] = self.select
-        start_date = self.get_starting_time(context).strftime("%Y-%m-%dT%H:%M:%S")
-        if self.replication_key:
-            date_filter = f"Modified gt datetime'{start_date}'"
+        start_date = self.get_starting_timestamp(context)
+        if start_date is not None:
+            date_filter = f"Modified gt datetime'{start_date.strftime('%Y-%m-%dT%H:%M:%S')}'"
             params["$filter"] = date_filter
         if next_page_token:
             params["$skiptoken"] = next_page_token
@@ -191,31 +180,3 @@ def post_process(
     def select(self) -> str:
         """Return the select query parameter."""
         return ",".join(self.schema["properties"].keys())
-
-
-class ExactSyncStream(ExactStream):
-    """Exact sync stream class."""
-
-    def get_new_paginator(self) -> BaseOffsetPaginator:
-        """Create a new pagination helper instance."""
-        return ExactPaginator(self, start_value=None, page_size=1000)
-
-    def get_starting_time(self, context: str) -> int:
-        """Return the starting timestamp for the stream."""
-        state = self.get_context_state(context)
-        rep_key = None
-        if "replication_key_value" in state:
-            rep_key = state["replication_key_value"]
-        return rep_key or 1
-
-    def get_url_params(self, context: dict | None, next_page_token: int) -> dict[str, Any]:
-        """Return a dictionary of parameters to use in the API request."""
-        params: dict = {}
-        if self.select:
-            params["$select"] = self.select
-        start_timestamp = self.get_starting_time(context)
-        date_filter = f"Timestamp gt {start_timestamp}" if start_timestamp == 1 else f"Timestamp gt {start_timestamp}L"
-        params["$filter"] = date_filter
-        if next_page_token:
-            params["$skiptoken"] = next_page_token
-        return params

tap_exact/schemas.py

@@ -11,7 +11,6 @@
 )
 
 transaction_lines_schema = PropertiesList(
-    Property("Timestamp", IntegerType),
     Property("Account", StringType),
     Property("AccountCode", StringType),
     Property("AccountName", StringType),
@@ -85,7 +84,6 @@
 ).to_dict()
 
 gl_accounts_schema = PropertiesList(
-    Property("Timestamp", IntegerType),
     Property("AllowCostsInSales", BooleanType),
     Property("AssimilatedVATBox", IntegerType),
     Property("BalanceSide", StringType),
@@ -131,7 +129,6 @@
 ).to_dict()
 
 gl_classifications_schema = PropertiesList(
-    Property("Timestamp", IntegerType),
     Property("Abstract", BooleanType),
     Property("Balance", StringType),
     Property("Code", StringType),

tap_exact/streams.py

@@ -4,7 +4,7 @@
 
 import typing as t
 
-from tap_exact.client import ExactStream, ExactSyncStream
+from tap_exact.client import ExactStream
 from tap_exact.schemas import (
     assets_schema,
     gl_account_classification_mappings_schema,
@@ -14,33 +14,32 @@
 )
 
 
-class TransactionLinesStream(ExactSyncStream):
+class TransactionLinesStream(ExactStream):
     """Define TransactionLines stream."""
 
     name = "transaction_lines"
-    path = "/sync/Financial/TransactionLines"
+    path = "/bulk/Financial/TransactionLines"
     primary_keys: t.ClassVar[list[str]] = ["ID", "Division"]
-    replication_key: t.ClassVar[str] = "Timestamp"
+    replication_key: t.ClassVar[str] = "Modified"
     schema = transaction_lines_schema  # pyright: ignore[reportAssignmentType]
 
 
-class GLAccountsStream(ExactSyncStream):
+class GLAccountsStream(ExactStream):
     """Define GLAccounts stream."""
 
     name = "gl_accounts"
-    path = "/sync/Financial/GLAccounts"
+    path = "/bulk/Financial/GLAccounts"
     primary_keys: t.ClassVar[list[str]] = ["ID", "Division"]
-    replication_key: t.ClassVar[str] = "Timestamp"
+    replication_key: t.ClassVar[str] = "Modified"
     schema = gl_accounts_schema  # pyright: ignore[reportAssignmentType]
 
 
-class GLClassificationsStream(ExactSyncStream):
+class GLClassificationsStream(ExactStream):
     """Define GLClassifications stream."""
 
     name = "gl_classifications"
     path = "/sync/Financial/GLClassifications"
     primary_keys: t.ClassVar[list[str]] = ["ID", "Division"]
-    replication_key: t.ClassVar[str] = "Timestamp"
     schema = gl_classifications_schema  # pyright: ignore[reportAssignmentType]