move to uv, update deps, fix tests (#49)

Author: tobiascadee

meltano.yml

@@ -14,7 +14,7 @@ plugins:
     - catalog
     - discover
     config:
-      start_date: 1741706134
+      start_date: 1751953786
       access_token: $TAP_INTERCOM_ACCESS_TOKEN
     select:
     - "*.*"

poetry.lock

@@ -1,59 +1,104 @@
-[tool.poetry]
+[project]
 name = "ticketswap-tap-intercom"
 version = "0.0.1"
 description = "`tap-intercom` is a Singer tap for Intercom, built with the Meltano Singer SDK."
 readme = "README.md"
-authors = ["Tobias Cadee <tobias.cadee@ticketswap.com>"]
+authors = [{ name = "Tobias Cadee", email = "tobias.cadee@ticketswap.com" }]
 keywords = [
     "ELT",
     "Intercom",
 ]
+classifiers = [
+    "Intended Audience :: Developers",
+    "License :: OSI Approved :: Apache Software License",
+    "Operating System :: OS Independent",
+    "Programming Language :: Python :: 3.9",
+    "Programming Language :: Python :: 3.10",
+    "Programming Language :: Python :: 3.11",
+    "Programming Language :: Python :: 3.12",
+    "Programming Language :: Python :: 3.13",
+]
 license = "Apache-2.0"
-packages = [
-    { include = "tap_intercom" },
+license-files = [ "LICENSE" ]
+requires-python = ">=3.9"
+
+dependencies = [
+    "singer-sdk>=0.42.1",
+    "fs-s3fs",
+    "requests",
+    "cached-property",  # Remove after Python 3.7 support is dropped
 ]
 
-[tool.poetry.dependencies]
-python = ">=3.8,<4"
-singer-sdk = { version="~=0.42.1" }
-fs-s3fs = { version = "~=1.1.1", optional = true }
-requests = "~=2.32.3"
-cached-property = "~=2" # Remove after Python 3.7 support is dropped
+[project.optional-dependencies]
+s3 = [
+    "s3fs",
+]
 
-[tool.poetry.group.dev.dependencies]
-pytest = ">=7.4.0"
-singer-sdk = { version="~=0.42.1", extras = ["testing"] }
+[project.scripts]
+# CLI declaration
+tap-intercom = 'tap_intercom.tap:TapIntercom.cli'
 
-[tool.poetry.extras]
-s3 = ["fs-s3fs"]
+[dependency-groups]
+dev = [
+    { include-group = "test" },
+]
+test = [
+    "pytest>=8",
+    "pytest-github-actions-annotate-failures>=0.3",
+    "singer-sdk[testing]",
+]
+[tool.hatch.build.targets.wheel]
+packages = [
+    "tap_intercom",
+]
+
+[tool.pytest.ini_options]
+addopts = [
+    "--durations=10",
+]
 
 [tool.mypy]
-python_version = "3.9"
 warn_unused_configs = true
+plugins = "sqlmypy"
 
-[tool.ruff]
+[tool.ruff.lint]
 ignore = [
-    "ANN101",  # missing-type-self
-    "ANN102",  # missing-type-cls
+    "COM812",  # missing-trailing-comma
 ]
 select = ["ALL"]
-src = ["tap_intercom"]
-target-version = "py37"
-
 
-[tool.ruff.flake8-annotations]
+[tool.ruff.lint.flake8-annotations]
 allow-star-arg-any = true
 
-[tool.ruff.isort]
-known-first-party = ["tap_intercom"]
-
-[tool.ruff.pydocstyle]
+[tool.ruff.lint.pydocstyle]
 convention = "google"
 
 [build-system]
-requires = ["poetry-core>=1.0.8"]
-build-backend = "poetry.core.masonry.api"
+requires = [
+    "hatchling>=1,<2",
+]
+build-backend = "hatchling.build"
 
-[tool.poetry.scripts]
-# CLI declaration
-tap-intercom = 'tap_intercom.tap:TapIntercom.cli'
+# This configuration can be used to customize tox tests as well as other test frameworks like flake8 and mypy
+[tool.tox]
+min_version = "4.22"
+requires = [
+    "tox>=4.22",
+    "tox-uv",
+]
+env_list = [
+    "py313",
+    "py312",
+    "py311",
+    "py310",
+    "py39",
+]
+
+[tool.tox.env_run_base]
+runner = "uv-venv-lock-runner"
+pass_env = [
+    "GITHUB_*",
+    "TAP_INTERCOM_*",
+]
+dependency_groups = [ "test" ]
+commands = [ [ "pytest", { replace = "posargs", default = [ "tests" ], extend = true } ] ]

pyproject.toml

@@ -53,7 +53,7 @@ def http_headers(self) -> dict:
 
     def get_url_params(self, context, next_page_token):
         params = {}
-        if self.rest_method == "GET":
+        if self.http_method == "GET":
             params = {"per_page": 150}
         return params
 
@@ -75,7 +75,7 @@ def prepare_request_payload(
             next_page_token: Token, page number or any request argument to request the
                 next page of data.
         """
-        if self.rest_method == "POST":
+        if self.http_method == "POST":
             body = {"sort": {"field": self.replication_key, "order": "ascending"}}
             start_date = self.get_starting_replication_key_value(context)
             if start_date or self.config.get("filters", {}).get(self.name):

tap_intercom/client.py

@@ -21,13 +21,14 @@ class ConversationsStream(IntercomStream):
     primary_keys: t.ClassVar[list[str]] = ["id"]
     replication_key = "updated_at"
     records_jsonpath = "$.conversations[*]"
-    rest_method = "POST"
+    http_method = "POST"
     schema = conversations_schema
 
     def get_child_context(self, record: dict, context: t.Optional[dict]) -> dict:
         """Return a context dictionary for child streams."""
         return {"conversation_id": record["id"]}
 
+
 class ConversationPartsStream(IntercomStream):
     name = "conversation_parts"
     parent_stream_type = ConversationsStream
@@ -42,32 +43,36 @@ def post_process(self, row: dict, context: dict | None = None) -> dict | None:
         row["conversation_id"] = context["conversation_id"]
         return row
 
+
 class AdminsStream(IntercomStream):
     name = "admins"
     path = "/admins"
     primary_keys: t.ClassVar[list[str]] = ["id"]
     records_jsonpath = "$.admins[*]"
     schema = admins_schema
 
+
 class TagsStream(IntercomStream):
     name = "tags"
     path = "/tags"
     primary_keys: t.ClassVar[list[str]] = ["id"]
     records_jsonpath = "$.data[*]"
     schema = tags_schema
 
+
 class TeamsStream(IntercomStream):
     name = "teams"
     path = "/teams"
     primary_keys: t.ClassVar[list[str]] = ["id"]
     records_jsonpath = "$.teams[*]"
     schema = teams_schema
 
+
 class ContactsStream(IntercomStream):
     name = "contacts"
     path = "/contacts/search"
     primary_keys: t.ClassVar[list[str]] = ["id"]
     records_jsonpath = "$.data[*]"
     replication_key = "updated_at"
-    rest_method = "POST"
+    http_method = "POST"
     schema = contacts_schema

tap_intercom/streams.py

@@ -3,20 +3,21 @@
 import datetime
 import os
 
-from singer_sdk.testing import get_tap_test_class
+from singer_sdk.testing import SuiteConfig, get_tap_test_class
 
 from tap_intercom.tap import TapIntercom
 
 SAMPLE_CONFIG = {
-    "start_date": (datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(hours=1)).strftime(
-        "%Y-%m-%dT%H:%M:%SZ"
-    ),
+    "start_date": int((datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(hours=1)).timestamp()),
     "access_token": os.getenv("TAP_INTERCOM_ACCESS_TOKEN"),
 }
 
+TEST_SUITE_CONFIG = SuiteConfig(max_records_limit=10, ignore_no_records=True)
+
 
 # Run standard built-in tap tests from the SDK:
 TestTapIntercom = get_tap_test_class(
     tap_class=TapIntercom,
     config=SAMPLE_CONFIG,
+    suite_config=TEST_SUITE_CONFIG,
 )