fully lint project with ruff (#254)

Author: tobiascadee

meltano.yml

@@ -16,14 +16,14 @@ plugins:
         - about
         - stream-maps
       config:
-        start_date: "2014-01-01T00:00:00Z"
-        client_id: $TAP_EXACT_CLIENT_ID_V2
-        client_secret: $TAP_EXACT_CLIENT_SECRET_V2
+        start_date: "2025-03-25T00:00:00Z"
+        client_id: $TAP_EXACT_CLIENT_ID
+        client_secret: $TAP_EXACT_CLIENT_SECRET
         tokens_s3_bucket: $TAP_EXACT_TOKENS_S3_BUCKET
         tokens_s3_key: $TAP_EXACT_TOKENS_S3_KEY
         divisions: ["3490573", "2542158", "2119843", "2140191", "2140277", "2603668"]
       select:
-        - gl_accounts.*
+        - *.*
 
   loaders:
     - name: target-jsonl

tap_exact/auth.py

@@ -2,17 +2,21 @@
 
 from __future__ import annotations
 
-from singer_sdk.authenticators import OAuthAuthenticator, SingletonMeta
-from singer_sdk.helpers._util import utc_now
-from singer_sdk.streams import RESTStream
-import backoff
-import requests
 import json
+from typing import TYPE_CHECKING
+
+import backoff
 import boto3
+import requests
+from singer_sdk.authenticators import OAuthAuthenticator, SingletonMeta
+from singer_sdk.helpers._util import utc_now
+
+if TYPE_CHECKING:
+    from singer_sdk.streams import RESTStream
 
 
 class EmptyResponseError(Exception):
-    """Raised when the response is empty"""
+    """Raised when the response is empty."""
 
 
 # The SingletonMeta metaclass makes your streams reuse the same authenticator instance.
@@ -35,6 +39,7 @@ def __init__(self, stream: RESTStream) -> None:
         self.access_token = tokens["access_token"]
 
     def download_tokens_from_s3(self) -> dict:
+        """Download tokens from S3."""
         return json.loads(
             self.s3.get_object(
                 Bucket=self._config["tokens_s3_bucket"], Key=self._config["tokens_s3_key"] + "/tokens.json"
@@ -44,6 +49,7 @@ def download_tokens_from_s3(self) -> dict:
         )
 
     def put_tokens_in_s3(self, tokens: dict) -> None:
+        """Put tokens in S3."""
         self.s3.put_object(
             Bucket=self._config["tokens_s3_bucket"],
             Key=self._config["tokens_s3_key"] + "/tokens.json",
@@ -66,6 +72,7 @@ def oauth_request_body(self) -> dict:
 
     @backoff.on_exception(backoff.expo, EmptyResponseError, max_tries=5, factor=2)
     def update_access_token(self) -> None:
+        """Update the access token using the refresh token."""
         request_time = utc_now()
         auth_request_payload = self.oauth_request_payload
         token_response = requests.post(
@@ -77,9 +84,10 @@ def update_access_token(self) -> None:
 
         try:
             if token_response.json().get("error_description") == "Rate limit exceeded: access_token not expired":
-                return None
+                return
         except Exception as e:
-            raise EmptyResponseError(f"Failed converting response to a json, because response is empty")
+            msg = "Failed converting response to a json, because response is empty"
+            raise EmptyResponseError(msg) from e
 
         try:
             token_response.raise_for_status()

tap_exact/client.py

@@ -2,24 +2,25 @@
 
 from __future__ import annotations
 
+import json
 import sys
-from pathlib import Path
-from typing import Any, Callable, Iterable, Optional, Dict
 import typing
-from datetime import datetime
+from pathlib import Path
+from typing import Any, Callable, Iterable
 
 import requests
-from singer_sdk.helpers.jsonpath import extract_jsonpath
-from singer_sdk.pagination import BaseOffsetPaginator  # noqa: TCH002
-from singer_sdk.streams import RESTStream
-from lxml import etree
-import json
 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")
@@ -34,11 +35,12 @@
 
 
 class ExactPaginator(BaseOffsetPaginator):
-    def __init__(self, stream, start_value, page_size) -> None:
+    """Exact pagination helper class."""
+    def __init__(self, stream:ExactStream, start_value:int, page_size:int) -> None:  # noqa: D107
         super().__init__(start_value=start_value, page_size=page_size)
         self.stream = stream
 
-    def has_more(self, response: Response) -> bool:  # noqa: ARG002
+    def has_more(self, response: Response) -> bool:
         """Override this method to check if the endpoint has any pages left.
 
         Args:
@@ -49,8 +51,9 @@ def has_more(self, response: Response) -> bool:  # noqa: ARG002
         """
         data = self.stream.xml_to_dict(response)
         link = data.get("feed", {}).get("link", [])
-        if type(link) == list:
+        if type(link) is list:
             return "next" in [item.get("@rel", "") for item in link]
+        return None
 
     def get_next(self, response: Response) -> TPageToken | None:
         """Get the next pagination token or index from the API response.
@@ -65,24 +68,26 @@ def get_next(self, response: Response) -> TPageToken | None:
         data = self.stream.xml_to_dict(response)
         link = data.get("feed", {}).get("link", [])
         if "next" in [item.get("@rel", "") for item in link]:
-            next_link = [item["@href"] for item in link if item["@rel"] == "next"][0]
-            next_page_token = next_link.split("&")[-1].split("=")[-1]
-            return next_page_token
+            next_link = next(item["@href"] for item in link if item["@rel"] == "next")
+            return next_link.split("&")[-1].split("=")[-1]
+        return None
 
 
 class ExactStream(RESTStream):
     """Exact stream class."""
 
     @property
     def partitions(self) -> list[dict] | None:
+        """Return a list of partitions, or None if the stream is not partitioned."""
         return [{"division": division} for division in self.config["divisions"]]
 
     @property
     def url_base(self) -> str:
         """Return the API URL root, configurable via tap settings."""
-        return f"https://start.exactonline.nl/api/v1"
+        return "https://start.exactonline.nl/api/v1"
 
     def get_url(self, context: dict | None) -> str:
+        """Return the URL for the API request."""
         return f"{self.url_base}/{context['division']}{self.path}"
 
     @cached_property
@@ -94,14 +99,16 @@ def authenticator(self) -> _Auth:
         """
         return ExactAuthenticator(self)
 
-    def get_starting_time(self, context):
+    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: Optional[dict], next_page_token) -> Dict[str, Any]:
+    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
@@ -117,15 +124,16 @@ def get_new_paginator(self) -> BaseOffsetPaginator:
         """Create a new pagination helper instance."""
         return ExactPaginator(self, start_value=None, page_size=60)
 
-    def xml_to_dict(self, response):
+    def xml_to_dict(self, response: requests.Response) -> dict:
+        """Convert xml response to dict."""
         try:
             # clean invalid xml characters
             my_parser = etree.XMLParser(recover=True)
-            xml = etree.fromstring(response.content, parser=my_parser)
+            xml = etree.fromstring(response.content, parser=my_parser)  # noqa: S320
             cleaned_xml_string = etree.tostring(xml)
             # parse xml to dict
             data = json.loads(json.dumps(xmltodict.parse(cleaned_xml_string)))
-        except:
+        except:  # noqa: E722
             data = json.loads(json.dumps(xmltodict.parse(response.content.decode("utf-8-sig").encode("utf-8"))))
         return data
 
@@ -159,7 +167,7 @@ def post_process(
         new_content = {}
         for key, value in content.items():
             new_key = key[2:]
-            if type(value) == str:
+            if type(value) is str:
                 new_content[new_key] = value
             elif not value or value.get("@m:null") == "true":
                 new_content[new_key] = None
@@ -176,11 +184,11 @@ def post_process(
                 new_content[new_key] = float(value.get("#text"))
             else:
                 new_content[new_key] = value.get("#text", None)
-        row = new_content
-        return row
+        return new_content
 
     @property
-    def select(self):
+    def select(self) -> str:
+        """Return the select query parameter."""
         return ",".join(self.schema["properties"].keys())
 
 
@@ -191,22 +199,21 @@ 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):
+    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.keys():
+        if "replication_key_value" in state:
             rep_key = state["replication_key_value"]
         return rep_key or 1
 
-    def get_url_params(self, context: Optional[dict], next_page_token) -> Dict[str, Any]:
+    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)
-        if start_timestamp == 1:
-            date_filter = f"Timestamp gt {start_timestamp}"
-        else:
-            date_filter = f"Timestamp gt {start_timestamp}L"
+        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

tap_exact/tap.py

@@ -3,9 +3,8 @@
 from __future__ import annotations
 
 from singer_sdk import Tap
-from singer_sdk.typing import DateTimeType, StringType, Property, PropertiesList, ArrayType
+from singer_sdk.typing import ArrayType, DateTimeType, PropertiesList, Property, StringType
 
-# TODO: Import your custom stream types here:
 from tap_exact import streams