Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 18 additions & 18 deletions core/change_detector.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,13 +6,13 @@

import os
from dataclasses import dataclass, field
from functools import lru_cache
from typing import Any, List, Optional
from enum import Enum
from functools import lru_cache
from typing import Any

from core.component_registry import BY_YAML_KEY, COMPONENT_TYPES
from core.normalization import is_explicit_list, normalize_values
from core.formatting import log_property_diffs
from core.normalization import is_explicit_list, normalize_values
from core.schema_reader import load_properties_for_type


Expand Down Expand Up @@ -150,7 +150,7 @@ class ComponentChange:
component_type: str # e.g., "interfaces", "power-ports"
component_name: str
change_type: ChangeType
property_changes: List[PropertyChange] = field(default_factory=list)
property_changes: list[PropertyChange] = field(default_factory=list)


@dataclass
Expand All @@ -161,9 +161,9 @@ class DeviceTypeChange:
model: str
slug: str
is_new: bool = False
property_changes: List[PropertyChange] = field(default_factory=list)
component_changes: List[ComponentChange] = field(default_factory=list)
netbox_id: Optional[int] = None
property_changes: list[PropertyChange] = field(default_factory=list)
component_changes: list[ComponentChange] = field(default_factory=list)
netbox_id: int | None = None

@property
def has_changes(self) -> bool:
Expand All @@ -180,8 +180,8 @@ def has_updates(self) -> bool:
class ChangeReport:
"""Aggregated change report for all device types."""

new_device_types: List[DeviceTypeChange] = field(default_factory=list)
modified_device_types: List[DeviceTypeChange] = field(default_factory=list)
new_device_types: list[DeviceTypeChange] = field(default_factory=list)
modified_device_types: list[DeviceTypeChange] = field(default_factory=list)
unchanged_count: int = 0


Expand Down Expand Up @@ -257,7 +257,7 @@ def __init__(self, device_types_instance, handle, remove_unmanaged_types: bool =
self.verbose = verbose
self.remove_unmanaged_types = remove_unmanaged_types

def detect_changes(self, device_types: List[dict], progress=None) -> ChangeReport:
def detect_changes(self, device_types: list[dict], progress=None) -> ChangeReport:
"""Analyze all device types and generate a change report.

Args:
Expand Down Expand Up @@ -308,7 +308,7 @@ def detect_changes(self, device_types: List[dict], progress=None) -> ChangeRepor

return report

def _compare_device_type_properties(self, yaml_data: dict, netbox_dt) -> List[PropertyChange]:
def _compare_device_type_properties(self, yaml_data: dict, netbox_dt) -> list[PropertyChange]:
"""Compare YAML device type properties against NetBox device type.

Args:
Expand Down Expand Up @@ -349,7 +349,7 @@ def _compare_device_type_properties(self, yaml_data: dict, netbox_dt) -> List[Pr
return changes

@staticmethod
def _compare_image_properties(yaml_data: dict, netbox_dt) -> List[PropertyChange]:
def _compare_image_properties(yaml_data: dict, netbox_dt) -> list[PropertyChange]:
"""Compare image properties between YAML and NetBox device type.

YAML uses boolean flags (front_image: true) meaning "an image should exist",
Expand Down Expand Up @@ -388,7 +388,7 @@ def _compare_components(
yaml_data: dict,
device_type_id: int,
parent_type: str = "device",
) -> List[ComponentChange]:
) -> list[ComponentChange]:
"""Compare all components between YAML and cached NetBox data.

Args:
Expand Down Expand Up @@ -418,9 +418,9 @@ def _compare_components(
# same as an empty list so chassis YAMLs that omit (e.g.) interfaces can
# still drive cleanup of stale templates in NetBox.
if yaml_key in yaml_data or self.remove_unmanaged_types:
for existing_name in existing_components.keys():
for existing_name in existing_components:
if existing_name not in yaml_component_names:
changes.append(
changes.append( # noqa: PERF401
ComponentChange(
component_type=yaml_key,
component_name=existing_name,
Expand Down Expand Up @@ -481,10 +481,10 @@ def _compare_component_properties(
self,
yaml_comp: dict,
netbox_comp,
properties: List[str],
properties: list[str],
comp_type: str = "",
manufacturer: str = "",
) -> List[PropertyChange]:
) -> list[PropertyChange]:
"""Compare properties between YAML and NetBox component.

*manufacturer* is the owning manufacturer's slug, used to resolve a relation
Expand Down Expand Up @@ -631,7 +631,7 @@ def _log_modified_summary(self, report: ChangeReport) -> None:
if parts:
self.handle.log(f" Breakdown: {', '.join(parts)}")

def _log_property_diffs(self, prop_changes: List[PropertyChange], indent: str) -> None:
def _log_property_diffs(self, prop_changes: list[PropertyChange], indent: str) -> None:
"""Emit diff-u style lines for *prop_changes* at the given *indent*."""
log_property_diffs(
[(pc.property_name, pc.old_value, pc.new_value) for pc in prop_changes],
Expand Down
2 changes: 1 addition & 1 deletion core/compat.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ def parse_netbox_version(version) -> tuple[int, int]:
suffixes NetBox ships ("4.7.0-beta2").
"""
raw = [int(re.sub(r"\D.*", "", part.strip()) or "0") for part in str(version).split(".")]
return tuple((raw + [0, 0])[:2]) # type: ignore[return-value]
return tuple(([*raw, 0, 0])[:2]) # type: ignore[return-value]


def supports_module_bay_types(version) -> bool:
Expand Down
7 changes: 2 additions & 5 deletions core/component_cache.py
Original file line number Diff line number Diff line change
Expand Up @@ -304,10 +304,7 @@ def get(self, endpoint_name, parent_type, parent_id, endpoint):
if key in cached:
return cached[key]

if parent_type == "device":
filter_kwargs = {"device_type_id": parent_id}
else:
filter_kwargs = {"module_type_id": parent_id}
filter_kwargs = {"device_type_id": parent_id} if parent_type == "device" else {"module_type_id": parent_id}
result = {item.name: item for item in endpoint.filter(**filter_kwargs)}
self.record(endpoint_name, parent_type, parent_id, result)
return result
Expand Down Expand Up @@ -378,7 +375,7 @@ def _finish_endpoint(self, endpoint_name, future):
"""Mark *endpoint_name* complete on the display, sizing the bar to the result."""
try:
total = max(len(future.result()), 1)
except Exception:
except Exception: # a worker error must not break the progress display # noqa: BLE001
total = 1
self._job["display"].finish(endpoint_name, total)
self._job["done"].add(endpoint_name)
Expand Down
3 changes: 1 addition & 2 deletions core/component_registry.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,6 @@
"""

from dataclasses import dataclass, field
from typing import Optional

# What a create call must resolve from a name to a NetBox id before it can POST.
LINK_BRIDGE = "bridge"
Expand Down Expand Up @@ -41,7 +40,7 @@ class ComponentType:
relations: tuple[str, ...] = field(default_factory=tuple)
graphql_extra: tuple[str, ...] = field(default_factory=tuple)
compare_extra: tuple[str, ...] = field(default_factory=tuple)
link: Optional[str] = None
link: str | None = None

@property
def graphql_fields(self):
Expand Down
38 changes: 19 additions & 19 deletions core/export.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,13 +3,15 @@
Entry point: ``Exporter(config, handle, export_dir, force_overwrite, vendor_slugs).run()``
"""

import contextlib
import hashlib
import os
import re
import threading
from collections.abc import Sequence
from dataclasses import dataclass
from pathlib import Path
from typing import Any, List, Optional, Sequence
from typing import Any

import requests
import yaml
Expand Down Expand Up @@ -108,7 +110,7 @@ class ExportItem:

kind: str # "device-type" | "module-type" | "rack-type"
nb_record: Any
repo_yaml: Optional[dict] # None when absent from repo
repo_yaml: dict | None # None when absent from repo
serialized: dict # What we will write
reason: str # "absent" | "differs" | "images-missing"
mfr_name: str
Expand Down Expand Up @@ -248,7 +250,7 @@ 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[Sequence[str]]):
def __init__(self, config, handle, export_dir: str, force_overwrite: bool, vendor_slugs: Sequence[str] | None):
"""Initialize the Exporter from the resolved run configuration."""
self.config = config
self.handle = handle
Expand All @@ -270,7 +272,7 @@ def __init__(self, config, handle, export_dir: str, force_overwrite: bool, vendo
handle=handle,
page_size=config.graphql_page_size,
)
self._module_image_details: Optional[dict] = None
self._module_image_details: dict | None = None

def _get_module_image_details(self) -> dict:
"""Return module type image details, fetching from NetBox at most once per run."""
Expand All @@ -297,7 +299,7 @@ def run(self, progress=None) -> None:
self.graphql.detect_module_bay_type_support()

# ── Fetch all types from NetBox ──────────────────────────────────────
by_model, by_slug = self.graphql.get_device_types(manufacturer_slugs=self.vendor_slugs)
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)

Expand Down Expand Up @@ -371,9 +373,9 @@ def _compare_vendors_to_items(
repo_dt_by_slug,
repo_mt_by_key,
progress,
) -> tuple[List[ExportItem], int]:
) -> tuple[list[ExportItem], int]:
"""Compare stale device/module types per vendor and return export items."""
items: List[ExportItem] = []
items: list[ExportItem] = []
skipped_fresh = 0
compare_task = (
progress.add_task("Comparing vendors", total=len(all_vendor_slugs))
Expand Down Expand Up @@ -441,9 +443,9 @@ def _compare_vendors_to_items(

return items, skipped_fresh

def _compare_racks_to_items(self, all_rt, manifest, repo_rt_by_key, progress) -> tuple[List[ExportItem], int]:
def _compare_racks_to_items(self, all_rt, manifest, repo_rt_by_key, progress) -> tuple[list[ExportItem], int]:
"""Compare stale rack types and return export items."""
items: List[ExportItem] = []
items: list[ExportItem] = []
skipped_fresh = 0
rack_records = [record for models in all_rt.values() for record in models.values()]
rack_task = (
Expand Down Expand Up @@ -698,10 +700,8 @@ def _fetch_one(endpoint_name):
results = list(pool.map(_fetch_one, COMPONENT_ENDPOINT_NAMES))
finally:
for client in _clients:
try:
with contextlib.suppress(Exception):
client.close()
except Exception:
pass

for endpoint_name, records in results:
for rec in records:
Expand All @@ -717,7 +717,7 @@ def _fetch_one(endpoint_name):

def _determine_export_set_for_device_types(
self, nb_records: list, repo_dt_by_slug: dict, components_by_dt_id: dict
) -> List[ExportItem]:
) -> list[ExportItem]:
"""Build the list of device types that need exporting to the repo.

Includes records absent from the repo, those whose serialized form
Expand All @@ -733,7 +733,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]
reason: str | None
if repo_yaml is None:
reason = "absent"
elif _repo_supersedes(repo_yaml, serialized):
Expand All @@ -759,7 +759,7 @@ def _determine_export_set_for_device_types(

def _determine_export_set_for_module_types(
self, nb_records: list, repo_mt_by_key: dict, components_by_mt_id: dict
) -> List[ExportItem]:
) -> list[ExportItem]:
"""Build the list of module types that need exporting to the repo.

Includes records absent from the repo and those whose serialized form
Expand Down Expand Up @@ -796,7 +796,7 @@ def _determine_export_set_for_module_types(
)
return items

def _determine_export_set_for_rack_types(self, nb_records: list, repo_rt_by_key: dict) -> List[ExportItem]:
def _determine_export_set_for_rack_types(self, nb_records: list, repo_rt_by_key: dict) -> list[ExportItem]:
"""Build the list of rack types that need exporting to the repo.

Includes records absent from the repo and those whose serialized form
Expand Down Expand Up @@ -833,7 +833,7 @@ def _determine_export_set_for_rack_types(self, nb_records: list, repo_rt_by_key:
)
return items

def _check_missing_images(self, front_url, rear_url, mfr_name: str, slug: str) -> Optional[str]:
def _check_missing_images(self, front_url, rear_url, mfr_name: str, slug: str) -> str | None:
"""Return ``'images-missing'`` if any expected local image is absent; else None.

DTL stores images under ``elevation-images/<Vendor>/<slug>.{front,rear}.{png,jpg,jpeg,gif}``
Expand Down Expand Up @@ -874,7 +874,7 @@ def _download_type_images(self, item: ExportItem) -> bool:
"""Download images for *item*. Returns True if all downloads succeeded."""
if item.kind == "device-type":
return self._download_device_type_images(item)
elif item.kind == "module-type":
if item.kind == "module-type":
return self._download_module_type_images(item)
return True # rack types have no images

Expand Down Expand Up @@ -975,7 +975,7 @@ def _download_module_type_images(self, item: ExportItem) -> bool:
return ok

def _download_image(
self, url_path: str, dest: Path, content_type_out: "Optional[list]" = None
self, url_path: str, dest: Path, content_type_out: "list | None" = None
) -> "str | _SkipSentinel | None":
"""Download an image from NetBox and write to *dest*.

Expand Down
6 changes: 4 additions & 2 deletions core/graphql_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,8 @@ def __getattr__(self, key):
try:
value = self[key]
except KeyError:
raise AttributeError(f"'DotDict' has no attribute '{key}'")
# The dict miss is an implementation detail of attribute lookup, not the error.
raise AttributeError(f"'DotDict' has no attribute '{key}'") from None
if isinstance(value, dict) and not isinstance(value, DotDict):
value = DotDict(value)
self[key] = value
Expand Down Expand Up @@ -114,7 +115,7 @@ def _response_body_detail(response):
return ""
try:
body = response.text.strip()
except Exception: # pragma: no cover - a body that cannot be decoded is not worth failing on
except Exception: # pragma: no cover - a body that cannot be decoded is not worth failing on # noqa: BLE001
return ""
if not body:
return ""
Expand Down Expand Up @@ -299,6 +300,7 @@ def query(self, graphql_query, variables=None, _retries=3):
raise GraphQLSchemaError(messages)

return body.get("data", {})
return None

def query_all(self, graphql_query, list_key, page_size=None, variables=None, on_page=None):
"""Auto-paginate a GraphQL list query using offset/limit.
Expand Down
Loading