chore(ty): migrate from mypy to ty

.github/workflows/workflow-linter.yml

@@ -84,6 +84,9 @@ jobs:
       - name: "Linting: ruff format"
         run: "uv run ruff format --check --diff ."
 
+      - name: "Linting: ty check"
+        run: "uv run ty check ."
+
       # TODO: Need to cleanup code
       # - name: "Pylint Tests"
       #   run: "uv run pylint infrahub_sync/**/*.py"

AGENTS.md

@@ -34,7 +34,6 @@ uv run infrahub-sync list --directory examples/
 # Make a change, then:
 uv run invoke format
 uv run invoke lint
-uv run mypy infrahub_sync/ --ignore-missing-imports
 uv run infrahub-sync list --directory examples/
 
 # If docs/CLI changed:
@@ -50,13 +49,14 @@ Run these in order before committing.
 uv sync
 uv run invoke format
 uv run invoke lint
-uv run mypy infrahub_sync/ --ignore-missing-imports
 ```
 
+`invoke lint` runs ruff → pylint → yamllint → ty.
+
 **Policy:**
 
 - New or changed code is Ruff-clean and typed where touched (docstrings, specific exceptions).
-- Do not increase existing mypy debt. If needed, use targeted `# type: ignore[<code>]` with a short TODO.
+- The codebase is clean under ty with no `[[tool.ty.overrides]]` blocks. Don't reintroduce overrides to mask type errors — fix the underlying issue, or use a targeted `# ty: ignore[<rule>]` with a short TODO at the call site.
 - If you add tests, run `uv run pytest -q`.
 
 **CLI sanity after changes:**
@@ -121,7 +121,7 @@ infrahub-sync/
 - Prefer explicit types on new or changed code.
 - Ruff: formatted and lint-clean. Honor `pyproject.toml`.
 - Pylint: fix actionable issues in touched code; some warnings are expected.
-- Mypy: run with `--ignore-missing-imports`; do not increase the error count.
+- ty: included in `uv run invoke lint`; do not increase the error count. For an ad-hoc check, `uv run ty check .` works too.
 - Public functions and classes require concise docstrings.
 - Raise specific exceptions; avoid broad `except Exception:`.
 
@@ -187,6 +187,7 @@ uv run invoke --list
 # linter.lint-ruff                Lint Python code with ruff
 # linter.lint-pylint              Lint Python code with pylint
 # linter.lint-yaml                Lint YAML files with yamllint
+# linter.lint-ty                  Type-check Python code with ty
 #
 # docs.*
 # docs.generate                   Generate CLI documentation
@@ -205,7 +206,7 @@ uv run invoke --list
 
 - Optional dependencies (for example, `pynetbox`, `pynautobot`) may be missing, producing import warnings.
 - `generate` and `sync` require running servers (Infrahub, NetBox, Nautobot).
-- Existing mypy debt exists; do not increase it and type the code you touch.
+- The codebase is clean under ty; there are no `[[tool.ty.overrides]]` blocks in `pyproject.toml`. Prefer real type fixes over reintroducing overrides.
 - Docs npm audit may flag dev-only vulnerabilities; they do not affect the Python package.
 
 ## Development Rules
@@ -215,7 +216,7 @@ uv run invoke --list
 - Do not force-push on shared branches.
 - Do not amend to hide pre-commit fixes; use a follow-up commit.
 - Apply PR labels: `bugs`, `breaking`, `enhancements`, `features` (default to `enhancements`).
-- Always run the required workflow (format → lint → mypy → CLI sanity) before a PR.
+- Always run the required workflow (format → lint → CLI sanity) before a PR. `invoke lint` now includes ty alongside ruff, pylint, and yamllint.
 
 ### Commit and PR Messages
 
@@ -236,7 +237,7 @@ uv run invoke --list
 **Approval checklist:**
 
 - [ ] Format and lint clean on changed areas.
-- [ ] No increase in mypy errors; new code typed.
+- [ ] `uv run ty check .` exits 0; new code typed.
 - [ ] CLI behaviors validated (`--help`, `list`, targeted `generate`).
 - [ ] Docs updated if flags or config changed.
 - [ ] Error handling uses specific exception types and clear messages.

infrahub_sync/__init__.py

@@ -3,9 +3,14 @@
 import logging
 import operator
 import re
-from typing import Any, Union
+from typing import TYPE_CHECKING, Any, ClassVar, Union
 
 import pydantic
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+    from diffsync.store import BaseStore
 from diffsync.enum import DiffSyncFlags
 from jinja2 import StrictUndefined
 from jinja2.nativetypes import NativeEnvironment
@@ -16,14 +21,13 @@
 
 logger = logging.getLogger(__name__)
 
+# Pydantic v1/v2 compatibility shim — runtime branch picks the right decorator + kwargs.
 if version.parse(pydantic.__version__) >= version.parse("2.0.0"):
-    # With Pydantic v2, we use `field_validator` with mode "before"
     from pydantic import field_validator as validator_decorator
 
-    validator_kwargs = {"mode": "before"}
+    validator_kwargs: dict[str, Any] = {"mode": "before"}
 else:
-    # With Pydantic v1, we use validator with `pre=True` and `allow_reuse=True`
-    from pydantic import validator as validator_decorator
+    from pydantic import validator as validator_decorator  # ty: ignore[deprecated]
 
     validator_kwargs = {"pre": True, "allow_reuse": True}
 
@@ -52,7 +56,7 @@ class SchemaMappingModel(pydantic.BaseModel):
     identifiers: list[str] | None = pydantic.Field(default=None)
     filters: list[SchemaMappingFilter] | None = pydantic.Field(default=None)
     transforms: list[SchemaMappingTransform] | None = pydantic.Field(default=None)
-    fields: list[SchemaMappingField] | None = []
+    fields: list[SchemaMappingField] = pydantic.Field(default_factory=list)
 
 
 class SyncAdapter(pydantic.BaseModel):
@@ -76,7 +80,7 @@ class SyncConfig(pydantic.BaseModel):
     schema_mapping: list[SchemaMappingModel] = []
     diffsync_flags: list[Union[str, DiffSyncFlags]] | None = []
 
-    @validator_decorator("diffsync_flags", **validator_kwargs)
+    @validator_decorator("diffsync_flags", **validator_kwargs)  # ty: ignore[no-matching-overload]
     def convert_str_to_enum(cls, v):
         if not isinstance(v, list):
             msg = "diffsync_flags must be provided as a list"
@@ -111,7 +115,7 @@ def convert_to_int(value: Any) -> int:
         raise ValueError(msg) from exc
 
 
-FILTERS_OPERATIONS = {
+FILTERS_OPERATIONS: dict[str, Callable[..., Any]] = {
     "==": operator.eq,
     "!=": operator.ne,
     ">": lambda field, value: operator.gt(convert_to_int(field), convert_to_int(value)),
@@ -131,6 +135,10 @@ def convert_to_int(value: Any) -> int:
 
 
 class DiffSyncMixin:
+    top_level: ClassVar[list[str]] = []
+    config: SyncConfig
+    store: BaseStore
+
     def load(self):
         """Load all the models, one by one based on the order defined in top_level."""
         for item in self.top_level:
@@ -146,6 +154,9 @@ def model_loader(self, model_name: str, model):
 
 
 class DiffSyncModelMixin:
+    # Set on generated subclasses (see generator/templates/diffsync_models.j2).
+    local_id: str | None = None
+
     @classmethod
     def apply_filter(cls, field_value: Any, operation: str, value: Any) -> bool:
         """Apply a specified operation to a field value."""
@@ -190,8 +201,9 @@ def apply_transform(cls, item: dict[str, Any], transform_expr: str, field: str)
             )
 
             # Allow subclasses to add custom filters
-            if hasattr(cls, "_add_custom_filters"):
-                cls._add_custom_filters(native_env, item)
+            add_custom_filters: Callable[..., None] | None = getattr(cls, "_add_custom_filters", None)
+            if add_custom_filters is not None:
+                add_custom_filters(native_env, item)
 
             # Compile the template with the native env
             template = native_env.from_string(transform_expr)
@@ -250,13 +262,18 @@ def get_resource_name(cls, schema_mapping: list[SchemaMappingModel]) -> str:
         """Get the resource name from the schema mapping."""
         for element in schema_mapping:
             if element.name == cls.__name__:
+                if element.mapping is None:
+                    msg = f"Resource mapping is unset for class {cls.__name__}"
+                    raise ValueError(msg)
                 return element.mapping
         msg = f"Resource name not found for class {cls.__name__}"
         raise ValueError(msg)
 
     @classmethod
     def is_list(cls, name):
-        field = cls.__fields__.get(name)
+        # Pydantic v2 exposes `model_fields`; v1 uses `__fields__`. Try both.
+        fields = getattr(cls, "model_fields", None) or getattr(cls, "__fields__", None) or {}
+        field = fields.get(name)
         if not field:
             msg = f"Unable to find the field {name} under {cls}"
             raise ValueError(msg)

infrahub_sync/adapters/aci.py

@@ -6,15 +6,15 @@
 import logging
 import os
 from datetime import datetime, timedelta, timezone
-from typing import TYPE_CHECKING, Any, ClassVar
+from typing import Any, ClassVar
 
-import requests  # type: ignore[import]
+import requests
 import urllib3
 from diffsync import Adapter, DiffSyncModel
-from requests import Response  # type: ignore[import]
-from requests.adapters import HTTPAdapter  # type: ignore[import]
+from requests import Response
+from requests.adapters import HTTPAdapter
 from typing_extensions import Self
-from urllib3.util.retry import Retry  # type: ignore[import]
+from urllib3.util.retry import Retry
 
 from infrahub_sync import (
     DiffSyncMixin,
@@ -26,9 +26,6 @@
 
 from .utils import get_value
 
-if TYPE_CHECKING:
-    import builtins
-
 logger = logging.getLogger(__name__)
 
 
@@ -233,7 +230,7 @@ def _create_aci_client(self, adapter: SyncAdapter) -> AciApiClient:
             verify=verify,
         )
 
-    def model_loader(self, model_name: str, model: builtins.type[AciModel]) -> None:  # type: ignore[valid-type]
+    def model_loader(self, model_name: str, model: type[AciModel]) -> None:
         """
         Load and process models using schema mapping filters and transformations.
 
@@ -270,25 +267,25 @@ def model_loader(self, model_name: str, model: builtins.type[AciModel]) -> None:
             is_source_adapter = self.config.source.name.lower() == "aci"
             if is_source_adapter:
                 # Filter records
-                filtered_objs = model.filter_records(records=objs, schema_mapping=element)  # type: ignore[attr-defined]
+                filtered_objs = model.filter_records(records=objs, schema_mapping=element)
                 logger.info("%s: Loading %d/%d %s", self.type, len(filtered_objs), total, resource_name)
                 # Transform records
-                transformed_objs = model.transform_records(records=filtered_objs, schema_mapping=element)  # type: ignore[attr-defined]
+                transformed_objs = model.transform_records(records=filtered_objs, schema_mapping=element)
             else:
                 logger.info("%s: Loading all %d %s", self.type, total, resource_name)
                 transformed_objs = objs
 
             # Create model instances after filtering and transforming
             for obj in transformed_objs:
                 data = self.obj_to_diffsync(obj=obj, mapping=element, model=model)
-                item = model(**data)  # type: ignore[misc]
+                item = model(**data)
                 self.add(item)
 
     def obj_to_diffsync(
         self,
         obj: dict[str, Any],
         mapping: SchemaMappingModel,
-        model: builtins.type[AciModel],  # type: ignore[valid-type]
+        model: type[AciModel],
     ) -> dict[str, Any]:
         """Convert an object to DiffSync format based on the provided mapping schema."""
         obj_id = self._extract_aci_id(obj)
@@ -298,7 +295,7 @@ def obj_to_diffsync(
         if not mapping.fields:
             return data
         for field in mapping.fields:
-            field_is_list = model.is_list(name=field.name)  # type: ignore[attr-defined]
+            field_is_list = model.is_list(name=field.name)
 
             if field.static:
                 data[field.name] = field.static

infrahub_sync/adapters/genericrestapi.py

@@ -1,16 +1,11 @@
 from __future__ import annotations
 
-import os
-from typing import TYPE_CHECKING, Any
-
-try:
-    from typing import Self
-except ImportError:
-    from typing_extensions import Self
-
 import logging
+import os
+from typing import Any
 
 from diffsync import Adapter, DiffSyncModel
+from typing_extensions import Self
 
 from infrahub_sync import (
     DiffSyncMixin,
@@ -25,9 +20,6 @@
 
 logger = logging.getLogger(__name__)
 
-if TYPE_CHECKING:
-    from collections.abc import Mapping
-
 
 class GenericrestapiAdapter(DiffSyncMixin, Adapter):
     """
@@ -134,7 +126,7 @@ def _create_rest_client(self, settings: dict) -> RestApiClient:
             verify=verify_ssl,
         )
 
-    def model_loader(self, model_name: str, model: GenericrestapiModel) -> None:
+    def model_loader(self, model_name: str, model: type[GenericrestapiModel]) -> None:
         """
         Load and process models using schema mapping filters and transformations.
 
@@ -165,7 +157,8 @@ def model_loader(self, model_name: str, model: GenericrestapiModel) -> None:
                 raise ValueError(msg) from exc
 
             total = len(objs)
-            if self.config.source.name.title() == self.type.title():
+            adapter_type_title = (self.type or "").title()
+            if self.config.source.name.title() == adapter_type_title:
                 # Filter records
                 filtered_objs = model.filter_records(records=objs, schema_mapping=element)
                 logger.info("%s: Loading %d/%d %s", self.type, len(filtered_objs), total, resource_name)
@@ -206,17 +199,16 @@ def _extract_objects_from_response(
         # Try to get data using the response key
         objs = response_data.get(response_key, response_data.get(resource_name, {}))
 
-        # Handle different response formats
+        # Filter each branch to dicts so `obj_to_diffsync` (which calls `.get(...)`) never sees non-dict items.
         if isinstance(objs, dict):
-            # If it's a dict, convert values to list (like Observium)
-            objs = list(objs.values())
-        elif not isinstance(objs, list):
-            # If it's neither dict nor list, wrap in list
-            objs = [objs] if objs else []
-
-        return objs
-
-    def obj_to_diffsync(self, obj: dict[str, Any], mapping: SchemaMappingModel, model: GenericrestapiModel) -> dict:
+            return [v for v in objs.values() if isinstance(v, dict)]
+        if isinstance(objs, list):
+            return [v for v in objs if isinstance(v, dict)]
+        return []
+
+    def obj_to_diffsync(
+        self, obj: dict[str, Any], mapping: SchemaMappingModel, model: type[GenericrestapiModel]
+    ) -> dict:
         """
         Convert an object to DiffSync format.
 
@@ -257,7 +249,7 @@ def obj_to_diffsync(self, obj: dict[str, Any], mapping: SchemaMappingModel, mode
                         if isinstance(node, dict):
                             matching_nodes = []
                             node_id = node.get("id", None)
-                            matching_nodes = [item for item in nodes if item.local_id == str(node_id)]
+                            matching_nodes = [item for item in nodes if item.local_id == str(node_id)]  # ty: ignore[unresolved-attribute]
                             if len(matching_nodes) == 0:
                                 msg = f"Unable to locate the node {model} {node_id}"
                                 raise IndexError(msg)
@@ -269,15 +261,15 @@ def obj_to_diffsync(self, obj: dict[str, Any], mapping: SchemaMappingModel, mode
 
                 else:
                     data[field.name] = []
-                    for node in get_value(obj, field.mapping):
+                    for node in get_value(obj, field.mapping) or []:
                         if not node:
                             continue
                         node_id = node.get("id", None)
                         if not node_id and isinstance(node, tuple):
                             node_id = node[1] if node[0] == "id" else None
                             if not node_id:
                                 continue
-                        matching_nodes = [item for item in nodes if item.local_id == str(node_id)]
+                        matching_nodes = [item for item in nodes if item.local_id == str(node_id)]  # ty: ignore[unresolved-attribute]
                         if len(matching_nodes) == 0:
                             msg = f"Unable to locate the node {field.reference} {node_id}"
                             raise IndexError(msg)
@@ -296,8 +288,8 @@ class GenericrestapiModel(DiffSyncModelMixin, DiffSyncModel):
     def create(
         cls,
         adapter: Adapter,
-        ids: Mapping[Any, Any],
-        attrs: Mapping[Any, Any],
+        ids: dict[Any, Any],
+        attrs: dict[Any, Any],
     ) -> Self | None:
         # TODO: To implement
         return super().create(adapter=adapter, ids=ids, attrs=attrs)