fix(export): accept vendor tuples in the GraphQL manufacturer filter

.github/workflows/tests.yml

@@ -55,5 +55,8 @@ jobs:
       - name: Check formatting
         run: uv run ruff format --check .
 
+      - name: Check types
+        run: uv run mypy
+
       - name: Check docstring coverage
         run: uv run interrogate

.pre-commit-config.yaml

@@ -19,6 +19,14 @@ repos:
         language: system
         types: [python]
         require_serial: true
+      # No filenames: mypy reads its own scope (core/ and nb-dt-import.py) from pyproject.toml.
+      - id: mypy
+        name: mypy
+        entry: uv run --native-tls mypy
+        language: system
+        types: [python]
+        pass_filenames: false
+        require_serial: true
       - id: interrogate
         name: interrogate
         entry: uv run --native-tls interrogate

core/change_detector.py

@@ -376,7 +376,7 @@ def _compare_component_properties(
                 # so that any change to rear port name, front_port_position, or
                 # rear_port_position is detected.
                 yaml_mappings = yaml_comp.get("_mappings") or []
-                yaml_set = frozenset(
+                yaml_set: frozenset = frozenset(
                     (
                         m.get("rear_port", ""),
                         m.get("front_port_position", 1),
@@ -392,7 +392,7 @@ def _compare_component_properties(
                 has_names = any(m.get("rear_port_name") is not None for m in canonical)
                 if has_names:
                     # NetBox >= 4.5: compare with rear port names
-                    netbox_set = frozenset(
+                    netbox_set: frozenset = frozenset(
                         (
                             m.get("rear_port_name", ""),
                             m.get("front_port_position", 1),

core/component_cache.py

@@ -12,6 +12,7 @@
 import concurrent.futures
 import queue
 import threading
+from typing import Any
 
 from core.compat import (
     device_type_filter_key,
@@ -136,9 +137,9 @@ def __init__(self, netbox, graphql, handle, new_filters, max_threads, wrap_recor
         self.max_threads = max_threads
         self._wrap_record = wrap_record or (lambda record: record)
 
-        self._entries = {}
+        self._entries: dict = {}
         self._ready = False
-        self._job = None
+        self._job: Any = None
 
     # ── State ────────────────────────────────────────────────────────────────
 
@@ -169,7 +170,7 @@ def begin_prefetch(self, manufacturer_slug=None, display=None):
         for component in COMPONENT_TYPES:
             display.add(component.endpoint, component.plural_label)
 
-        updates = queue.Queue()
+        updates: queue.Queue = queue.Queue()
         max_workers = max(1, min(len(COMPONENT_TYPES), self.max_threads))
         executor = concurrent.futures.ThreadPoolExecutor(max_workers=max_workers)
         worker_state = threading.local()
@@ -369,7 +370,7 @@ def _pending(self):
 
     def _drain_updates(self):
         """Apply every queued page count to the display, coalescing per endpoint."""
-        totals = {}
+        totals: dict = {}
         while True:
             try:
                 endpoint_name, advance = self._job["updates"].get_nowait()

core/component_registry.py

@@ -25,10 +25,10 @@ class ComponentType:
     yaml_key: str
     endpoint: str
     label: str
-    fields: tuple
+    fields: tuple[str, ...]
     module_types: bool = True
-    graphql_extra: tuple = field(default_factory=tuple)
-    compare_extra: tuple = field(default_factory=tuple)
+    graphql_extra: tuple[str, ...] = field(default_factory=tuple)
+    compare_extra: tuple[str, ...] = field(default_factory=tuple)
     link: Optional[str] = None
 
     @property

core/config.py

@@ -56,8 +56,8 @@ class RunConfig:
     repo_branch: str
     repo_path: str
 
-    vendors: tuple = ()
-    slugs: tuple = ()
+    vendors: tuple[str, ...] = ()
+    slugs: tuple[str, ...] = ()
 
     export_diff: bool = False
     export_diff_dir: str = "extra/"
@@ -74,7 +74,7 @@ class RunConfig:
     show_remaining_time: bool = False
 
     # Resolution decisions worth telling the user about, logged once the handler exists.
-    notices: tuple = field(default_factory=tuple)
+    notices: tuple[str, ...] = field(default_factory=tuple)
 
 
 def _text(env, name, default=None):
@@ -246,7 +246,7 @@ def _split_slugs(values):
     return tuple(s.strip() for slug in values for s in slug.split(",") if s.strip())
 
 
-def resolve_run_config(argv=None, env=None):
+def resolve_run_config(argv=None, env=None) -> RunConfig:
     """Resolve *argv* and *env* into one RunConfig, or raise ConfigError.
 
     Reads argv before the environment, so ``--help`` answers without an environment

core/export.py

@@ -9,7 +9,7 @@
 import threading
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any, List, Optional
+from typing import Any, List, Optional, Sequence
 
 import requests
 import yaml
@@ -30,7 +30,12 @@
 from core.netbox_api import IMAGE_EXTENSIONS, _build_auth_header
 from core.repo import LIBRARY_TYPE_DIRS, library_dirs_present
 
-_SKIP = object()  # sentinel: image already exists, no download needed
+
+class _SkipSentinel:
+    """Type of the ``_SKIP`` sentinel: the image already exists, so no download is needed."""
+
+
+_SKIP = _SkipSentinel()
 
 # Maps Content-Type to a canonical extension for extension-less attachments.
 _CONTENT_TYPE_EXT = {
@@ -191,13 +196,17 @@ def _is_subset(sub: Any, sup: Any) -> bool:
 class Exporter:
     """Exports NetBox device/module/rack types to a local directory in DTL format."""
 
-    def __init__(self, config, handle, export_dir: str, force_overwrite: bool, vendor_slugs: Optional[List[str]]):
+    def __init__(self, config, handle, export_dir: str, force_overwrite: bool, vendor_slugs: Optional[Sequence[str]]):
         """Initialize the Exporter from the resolved run configuration."""
         self.config = config
         self.handle = handle
         self.export_dir = Path(export_dir)
         self.force_overwrite = force_overwrite
-        self.vendor_slugs = vendor_slugs  # None means all vendors
+        # tuple("cisco") is five single-character slugs, each one valid to the GraphQL layer.
+        if isinstance(vendor_slugs, (str, bytes)):
+            raise ValueError("vendor_slugs must be None or a sequence of vendor slugs, not a bare string")
+        # None means all vendors. The GraphQL layer rejects an empty sequence, so normalize one here.
+        self.vendor_slugs = tuple(vendor_slugs) if vendor_slugs else None
         self.repo_path = Path(config.repo_path)
         self.base_url = config.netbox_url.rstrip("/")
         self.token = config.netbox_token
@@ -233,11 +242,9 @@ def run(self, progress=None) -> None:
         self.handle.log(f"Export-diff: fetching NetBox device/module/rack types{scope}")
 
         # ── Fetch all types from NetBox ──────────────────────────────────────
-        by_model, by_slug = self.graphql.get_device_types(
-            manufacturer_slugs=self.vendor_slugs if self.vendor_slugs else None
-        )
-        all_mt = self.graphql.get_module_types(manufacturer_slugs=self.vendor_slugs if self.vendor_slugs else None)
-        all_rt = self.graphql.get_rack_types(manufacturer_slugs=self.vendor_slugs if self.vendor_slugs else None)
+        by_model, by_slug = self.graphql.get_device_types(manufacturer_slugs=self.vendor_slugs)
+        all_mt = self.graphql.get_module_types(manufacturer_slugs=self.vendor_slugs)
+        all_rt = self.graphql.get_rack_types(manufacturer_slugs=self.vendor_slugs)
 
         total_dt = len(by_model)
         total_mt = sum(len(v) for v in all_mt.values())
@@ -671,6 +678,7 @@ def _determine_export_set_for_device_types(
             manifest_key = f"{mfr_name}/{rec.slug}"
 
             repo_yaml = repo_dt_by_slug.get((mfr_slug, rec.slug))
+            reason: Optional[str]
             if repo_yaml is None:
                 reason = "absent"
             elif _repo_supersedes(repo_yaml, serialized):
@@ -911,7 +919,9 @@ def _download_module_type_images(self, item: ExportItem) -> bool:
                 ok = False
         return ok
 
-    def _download_image(self, url_path: str, dest: Path, content_type_out: "Optional[list]" = None) -> "Optional[str]":
+    def _download_image(
+        self, url_path: str, dest: Path, content_type_out: "Optional[list]" = None
+    ) -> "str | _SkipSentinel | None":
         """Download an image from NetBox and write to *dest*.
 
         Returns SHA-256 hex digest on success, None on failure, or the module-level

core/export_manifest.py

@@ -9,7 +9,7 @@
 import os
 from pathlib import Path
 
-_EMPTY = {"device-types": {}, "module-types": {}, "rack-types": {}}
+_EMPTY: dict = {"device-types": {}, "module-types": {}, "rack-types": {}}
 
 
 def load_manifest(path: Path) -> dict: