chore: Fix typing (#266)

.github/workflows/ci_workflow.yml

@@ -54,3 +54,6 @@ jobs:
          TAP_FACEBOOK_START_DATE: ${{ secrets.tap_facebook_start_date }}
       run: |
         uvx --with=tox-uv tox -e ${{ matrix.python-version }}
+    - name: Check types
+      run: |
+        uvx --with=tox-uv tox -e typing

.pre-commit-config.yaml

@@ -27,11 +27,3 @@ repos:
   - id: ruff
     args: ["--fix"]
   - id: ruff-format
-
-- repo: https://github.com/pre-commit/mirrors-mypy
-  rev: v1.13.0
-  hooks:
-  - id: mypy
-    pass_filenames: true
-    additional_dependencies:
-    - types-requests

pyproject.toml

@@ -24,6 +24,10 @@ singer-sdk = "~=0.42.0"
 pytest = ">=8.2"
 singer-sdk = {version = "~=0.42.0", extras = ["testing"]}
 
+[tool.poetry.group.typing.dependencies]
+mypy = ">=1.13"
+types-requests = "~=2.32"
+
 [tool.isort]
 profile = "black"
 multi_line_output = 3 # Vertical Hanging Indent
@@ -59,3 +63,15 @@ typing = "t"
 
 [tool.ruff.lint.pydocstyle]
 convention = "google"
+
+[tool.mypy]
+warn_redundant_casts = true
+warn_unreachable = true
+warn_unused_configs = true
+warn_unused_ignores = true
+
+[[tool.mypy.overrides]]
+ignore_missing_imports = true
+module = [
+    "facebook_business.*", # TODO: Remove when https://github.com/facebook/facebook-python-business-sdk/issues/657 is shipped
+]

tap_facebook/client.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import abc
 import json
 import typing as t
 from http import HTTPStatus
@@ -15,6 +16,7 @@
 
 if t.TYPE_CHECKING:
     import requests
+    from singer_sdk.helpers.types import Context
 
 
 class FacebookStream(RESTStream):
@@ -71,7 +73,7 @@ def get_next_page_token(
 
     def get_url_params(
         self,
-        context: dict | None,  # noqa: ARG002
+        context: Context | None,  # noqa: ARG002
         next_page_token: t.Any | None,  # noqa: ANN401
     ) -> dict[str, t.Any]:
         """Return a dictionary of values to be used in URL parameterization.
@@ -143,10 +145,15 @@ def backoff_max_tries(self) -> int:
         return 20
 
 
-class IncrementalFacebookStream(FacebookStream):
+class IncrementalFacebookStream(FacebookStream, metaclass=abc.ABCMeta):
+    @property
+    @abc.abstractmethod
+    def filter_entity(self) -> str:
+        """The entity to filter on."""
+
     def get_url_params(
         self,
-        context: dict | None,
+        context: Context | None,
         next_page_token: t.Any | None,  # noqa: ANN401
     ) -> dict[str, t.Any]:
         """Return a dictionary of values to be used in URL parameterization.
@@ -164,13 +171,13 @@ def get_url_params(
         if self.replication_key:
             params["sort"] = "asc"
             params["order_by"] = self.replication_key
-            ts = pendulum.parse(self.get_starting_replication_key_value(context))
+            ts = pendulum.parse(self.get_starting_replication_key_value(context))  # type: ignore[arg-type]
             params["filtering"] = json.dumps(
                 [
                     {
                         "field": f"{self.filter_entity}.{self.replication_key}",
                         "operator": "GREATER_THAN",
-                        "value": int(ts.timestamp()),
+                        "value": int(ts.timestamp()),  # type: ignore[union-attr]
                     },
                 ],
             )

tap_facebook/streams/ad_accounts.py

@@ -17,6 +17,9 @@
 
 from tap_facebook.client import FacebookStream
 
+if t.TYPE_CHECKING:
+    from singer_sdk.helpers.types import Context, Record
+
 
 class AdAccountsStream(FacebookStream):
     """https://developers.facebook.com/docs/graph-api/reference/user/accounts/."""
@@ -214,9 +217,9 @@ def url_base(self) -> str:
 
     def post_process(
         self,
-        row: dict,
-        context: dict | None = None,  # noqa: ARG002
-    ) -> dict | None:
+        row: Record,
+        context: Context | None = None,  # noqa: ARG002
+    ) -> Record | None:
         row["amount_spent"] = int(row["amount_spent"]) if "amount_spent" in row else None
         row["balance"] = int(row["balance"]) if "balance" in row else None
         row["min_campaign_group_spend_cap"] = (
@@ -229,7 +232,7 @@ def post_process(
 
     def get_url_params(
         self,
-        context: dict | None,  # noqa: ARG002
+        context: Context | None,  # noqa: ARG002
         next_page_token: t.Any | None,  # noqa: ANN401
     ) -> dict[str, t.Any]:
         """Return a dictionary of values to be used in URL parameterization.

tap_facebook/streams/ad_insights.py

@@ -17,6 +17,9 @@
 from singer_sdk import typing as th
 from singer_sdk.streams.core import REPLICATION_INCREMENTAL, Stream
 
+if t.TYPE_CHECKING:
+    from singer_sdk.helpers.types import Context
+
 EXCLUDED_FIELDS = [
     "total_postbacks",
     "adset_end",
@@ -75,7 +78,7 @@ def __init__(self, *args, **kwargs) -> None:  # noqa: ANN002, ANN003
         super().__init__(*args, **kwargs)
 
     @property
-    def primary_keys(self) -> list[str] | None:
+    def primary_keys(self) -> t.Sequence[str] | None:
         return ["date_start", "account_id", "ad_id"] + self._report_definition["breakdowns"]
 
     @primary_keys.setter
@@ -88,11 +91,12 @@ def primary_keys(self, new_value: list[str] | None) -> None:
         self._primary_keys = new_value
 
     @staticmethod
-    def _get_datatype(field: str) -> th.Type | None:
+    def _get_datatype(field: str) -> th.JSONTypeHelper | None:
         d_type = AdsInsights._field_types[field]  # noqa: SLF001
         if d_type == "string":
             return th.StringType()
         if d_type.startswith("list"):
+            sub_props: list[th.Property]
             if "AdsActionStats" in d_type:
                 sub_props = [
                     th.Property(field.replace("field_", ""), th.StringType())
@@ -122,14 +126,21 @@ def _get_datatype(field: str) -> th.Type | None:
     @property
     @lru_cache  # noqa: B019
     def schema(self) -> dict:
-        properties: th.List[th.Property] = []
+        properties: list[th.Property] = []
         columns = list(AdsInsights.Field.__dict__)[1:]
         for field in columns:
             if field in EXCLUDED_FIELDS:
                 continue
-            properties.append(th.Property(field, self._get_datatype(field)))
-        for breakdown in self._report_definition["breakdowns"]:
-            properties.append(th.Property(breakdown, th.StringType()))
+            if data_type := self._get_datatype(field):
+                properties.append(th.Property(field, data_type))
+
+        properties.extend(
+            [
+                th.Property(breakdown, th.StringType())
+                for breakdown in self._report_definition["breakdowns"]
+            ],
+        )
+
         return th.PropertiesList(*properties).to_dict()
 
     def _initialize_client(self) -> None:
@@ -146,7 +157,7 @@ def _initialize_client(self) -> None:
             msg = f"Couldn't find account with id {account_id}"
             raise RuntimeError(msg)
 
-    def _run_job_to_completion(self, params: dict) -> th.Any:
+    def _run_job_to_completion(self, params: dict) -> None:
         job = self.account.get_insights(
             params=params,
             is_async=True,
@@ -212,13 +223,13 @@ def _get_selected_columns(self) -> list[str]:
 
     def _get_start_date(
         self,
-        context: dict | None,
+        context: Context | None,
     ) -> pendulum.Date:
         lookback_window = self._report_definition["lookback_window"]
 
-        config_start_date = pendulum.parse(self.config["start_date"]).date()
-        incremental_start_date = pendulum.parse(
-            self.get_starting_replication_key_value(context),
+        config_start_date = pendulum.parse(self.config["start_date"]).date()  # type: ignore[union-attr]
+        incremental_start_date = pendulum.parse(  # type: ignore[union-attr]
+            self.get_starting_replication_key_value(context),  # type: ignore[arg-type]
         ).date()
         lookback_start_date = incremental_start_date.subtract(days=lookback_window)
 
@@ -255,13 +266,13 @@ def _get_start_date(
 
     def get_records(
         self,
-        context: dict | None,
+        context: Context | None,
     ) -> t.Iterable[dict | tuple[dict, dict | None]]:
         self._initialize_client()
 
         time_increment = self._report_definition["time_increment_days"]
 
-        sync_end_date = pendulum.parse(
+        sync_end_date = pendulum.parse(  # type: ignore[union-attr]
             self.config.get("end_date", pendulum.today().to_date_string()),
         ).date()
 
@@ -287,7 +298,7 @@ def get_records(
                     "until": report_end.to_date_string(),
                 },
             }
-            job = self._run_job_to_completion(params)
+            job = self._run_job_to_completion(params)  # type: ignore[func-returns-value]
             for obj in job.get_result():
                 yield obj.export_all_data()
             # Bump to the next increment

tap_facebook/streams/campaign.py

@@ -2,11 +2,16 @@
 
 from __future__ import annotations
 
+import typing as t
+
 from singer_sdk import typing as th  # JSON Schema typing helpers
 from singer_sdk.streams.core import REPLICATION_INCREMENTAL
 
 from tap_facebook.client import IncrementalFacebookStream
 
+if t.TYPE_CHECKING:
+    from singer_sdk.helpers.types import Context, Record
+
 
 class CampaignStream(IncrementalFacebookStream):
     """https://developers.facebook.com/docs/marketing-api/reference/ad-campaign-group."""
@@ -133,9 +138,9 @@ class CampaignStream(IncrementalFacebookStream):
 
     def post_process(
         self,
-        row: dict,
-        context: dict | None,  # noqa: ARG002
-    ) -> dict:
+        row: Record,
+        context: Context | None = None,  # noqa: ARG002
+    ) -> Record | None:
         daily_budget = row.get("daily_budget")
         row["daily_budget"] = int(daily_budget) if daily_budget is not None else None
         return row