Add claude.ai fix (WIP)

pyproject.toml

@@ -1,3 +1,7 @@
+[project]
+name = "tap-linear"
+requires-python = ">=3.11"  # server constraint
+
 [tool.poetry]
 name = "leukeleu-tap-linear"
 version = "0.1.0"
@@ -29,31 +33,120 @@ python_version = "3.9"
 warn_unused_configs = true
 
 [tool.ruff]
+unsafe-fixes = false
+line-length = 88
+extend-exclude = [
+
+  "*/migrations/*",
+]
+
+[tool.ruff.lint]
+select = [
+  "ALL",  # enable all rules
+]
+preview = true  # even the experimental ones
 ignore = [
-    "ANN101",  # missing-type-self
-    "ANN102",  # missing-type-cls
+  "A003",  # builtins may be shadowed by class atrributes (e.g. "id")
+  "ANN",  # annotations are not (yet) required
+  "ARG",  # unused arguments are allowed
+  "COM812",  # trailing commas in function calls are not required
+  "COM819",  # trailing comma in tuple is allowed
+  "CPY001",  # copyright notice is not required
+  "D",  # docstring checks are disabled
+  "DJ008",  # __str__ methods are not required for models
+  "EM",  # error messages do not need to be assigned to a variable
+  "FIX",  # tasks/issue can be fixed later
+  "FURB101",  # open/read is OK, pathlib is not required
+  "FURB103",  # open/write is OK, pathlib is not required
+  "FURB118",  # operator module is nice, but not required
+  "PLC1901",  # `foo == ""` is *not equivalent* to `not foo`
+  "PT",  # pytest style is not required
+  "PTH",  # use of pathlib is not required
+  "PYI",  # disable type hinting stub file checks
+  "RET503",  # explicit return of None is not required
+  "RET505",  # else after return is OK
+  "RET506",  # else after raise is OK
+  "RUF012",  # typing.ClassVar is not required
+  "SIM105",  # contextlib.suppress and try/except/pass are both OK
+  "SIM108",  # if/else blocks and tenary operators are both OK
+  "TD001",  # "FIXME" is OK
+  "TD002",  # task author is not required
+  "TD003",  # link to issue is recommended but not required
+  "TID252",  # relative imports are OK
+  "TRY003",  # long error messages are OK
+]
+allowed-confusables = [
+  # Allow some confusable characters, e.g.:
+  # "–",  # (EN DASH)
 ]
-select = ["ALL"]
-src = ["tap_linear"]
-target-version = "py38"
 unfixable = [
-    "ERA001",  # commented out code
+  # manually fix/noqa
+  "RUF100",  # unused/unknown noqa comments
 ]
 
+[tool.ruff.lint.extend-per-file-ignores]
+"tests/*" = [
+  # This is OK in test code
+  "PLR0904",  # too many public methods
+  "PLR2004",  # magic values
+  "PLR6301",  # methods do not have to use self
+  "S105", "S106", "S107",  # hardcoded passwords
+  "S311",  # pseudo-random number generators
+]
 
-[tool.ruff.flake8-annotations]
-allow-star-arg-any = true
+[tool.ruff.lint.flake8-self]
+extend-ignore-names = [
+  "_default_manager",  # django
+  "_meta",  # django
+]
 
-[tool.ruff.isort]
-known-first-party = ["tap_linear"]
+[tool.ruff.lint.mccabe]
+max-complexity = 12
+
+[tool.ruff.lint.pep8-naming]
+extend-ignore-names = [
+  # add custom camel-case names (e.g. assertFooBar)
+]
 
-[tool.ruff.pydocstyle]
-convention = "google"
+[tool.ruff.lint.pycodestyle]
+max-doc-length = 88
+
+[tool.ruff.lint.pylint]
+max-args = 10  # default is 5
+
+[tool.ruff.lint.isort]
+lines-between-types = 1
+section-order = [
+  "future",
+  "standard-library",
+  "third-party",
+  "django",
+  "first-party",
+  "local-folder",
+]
+
+[tool.ruff.lint.isort.sections]
+django = [
+  "django",
+]
+
+[tool.coverage.run]
+branch = true
+source = [
+  "tap-linear",
+  "tests",
+]
+omit = [
+
+  "**/[aw]sgi.py",
+  "**/migrations/*",
+  "**/settings.py",
+]
 
-[build-system]
-requires = ["poetry-core>=1.0.8"]
-build-backend = "poetry.core.masonry.api"
+[tool.coverage.report]
+show_missing = true
+skip_covered = true
+skip_empty = true
 
-[tool.poetry.scripts]
-# CLI declaration
-tap-linear = 'tap_linear.tap:TapLinear.cli'
+[tool.coverage.html]
+directory = "../var/htmlcov"

tap_linear/client.py

@@ -1,10 +1,12 @@
-"""GraphQL client handling, including LinearStream base class."""
-
 from __future__ import annotations
 
-from typing import Any
+import copy
+import logging
+
+from typing import Any, Dict, Iterable, Optional
 
 from singer_sdk.authenticators import APIKeyAuthenticator
+from singer_sdk.exceptions import FatalAPIError
 from singer_sdk.helpers._classproperty import classproperty
 from singer_sdk.streams import GraphQLStream
 
@@ -18,31 +20,17 @@ class LinearStream(GraphQLStream):
     def authenticator(self) -> APIKeyAuthenticator:
         """Return a new authenticator object."""
         return APIKeyAuthenticator(
-            self,
-            key="Authorization",
-            value=self.config.get("auth_token"),
+            self, key="Authorization", value=self.config.get("auth_token")
         )
 
     @property
     def url_base(self) -> str:
         """Return the API URL root, configurable via tap settings."""
         return self.config.get("api_url")
 
-    @classproperty
-    def records_jsonpath(cls) -> str:  # noqa: N805
-        """Return the JSON path to the list of records for the stream."""
-        return f"$.data.{cls.name}.nodes[*]"
-
-    @classproperty
-    def next_page_token_jsonpath(cls) -> str:  # noqa: N805
-        """Return the JSON path to the next page token for the stream."""
-        return f"$.data.{cls.name}.pageInfo.endCursor"
-
     def get_url_params(
-        self,
-        context: dict | None,
-        next_page_token: str | None,
-    ) -> dict[str, Any] | str:
+        self, context: Optional[dict], next_page_token: Optional[Any]
+    ) -> Dict[str, Any]:
         """Return the URL params needed.
 
         These URL params are actually GraphQL variables.
@@ -54,21 +42,83 @@ def get_url_params(
         Returns:
             A dictionary of URL params.
         """
-        params = {"next": next_page_token}
-
-        if starting_timestamp := self.get_starting_timestamp(context):
-            replication_key_value = starting_timestamp.strftime("%Y-%m-%dT%H:%M:%SZ")
-            params["replicationKeyValue"] = replication_key_value
-
-        return params
-
-    def post_process(
-        self,
-        row: dict,
-        context: dict | None = None,  # noqa: ARG002
-    ) -> dict | None:
-        """Post-process row.
-
-        Flatten nested nodes lists.
-        """
-        return flatten_node_lists(row)
+        params: Dict[str, Any] = {}
+
+        if next_page_token:
+            params["after"] = next_page_token
+
+        if self.replication_key and (
+            starting_timestamp := self.get_starting_timestamp(context)
+        ):
+            params["filter"] = {
+                self.replication_key: {"gt": starting_timestamp.isoformat()}
+            }
+
+        return {"variables": params}
+
+    def prepare_request_payload(
+        self, context: Optional[dict], next_page_token: Optional[Any]
+    ) -> Dict[str, Any]:
+        query = self.get_query(context)
+        variables = self.get_url_params(context, next_page_token)["variables"]
+        return {
+            "query": query,
+            "variables": variables,
+        }
+
+    def parse_response(self, response: dict) -> Iterable[dict]:
+        try:
+            data = response["data"][self.name]["nodes"]
+            for row in data:
+                yield self.post_process(row)
+        except KeyError as e:
+            logging.error(f"Unexpected response structure: {response}")
+            logging.error(f"KeyError: {e}")
+            raise
+
+    def post_process(self, row: dict, context: Optional[dict] = None) -> Optional[dict]:
+        try:
+            return flatten_node_lists(row)
+        except Exception as e:
+            logging.error(f"Error in post_process: {e}")
+            logging.error(f"Problematic row: {row}")
+            return None
+
+    def get_next_page_token(
+        self, response: Dict[str, Any], previous_token: Optional[Any] = None
+    ) -> Optional[Any]:
+        try:
+            has_next_page = response["data"][self.name]["pageInfo"]["hasNextPage"]
+            if has_next_page:
+                return response["data"][self.name]["pageInfo"]["endCursor"]
+        except KeyError:
+            logging.error(f"Unexpected response structure for pagination: {response}")
+        return None
+
+    def request_records(self, context: Optional[dict]) -> Iterable[dict]:
+        next_page_token: Any = None
+        finished = False
+        while not finished:
+            prepared_request = self.prepare_request(
+                context, next_page_token=next_page_token
+            )
+            resp = self._request(prepared_request, context)
+            for row in self.parse_response(resp):
+                yield row
+            previous_token = copy.deepcopy(next_page_token)
+            next_page_token = self.get_next_page_token(resp, previous_token)
+            if next_page_token and next_page_token == previous_token:
+                raise RuntimeError(
+                    f"Loop detected in pagination. Token {next_page_token} is identical to prior token."
+                )
+            finished = not next_page_token
+
+    def _request(
+        self, prepared_request, context: Optional[dict] = None
+    ) -> Dict[str, Any]:
+        response = self.requests_session.send(prepared_request)
+        if response.status_code != 200:
+            raise FatalAPIError(
+                f"Request failed with status {response.status_code}: {response.text}"
+            )
+        return response.json()