fix(types): resolve all 427 mypy errors

opennms_api_wrapper/_acks.py

@@ -1,7 +1,10 @@
 """Acknowledgements REST API – /rest/acks."""
+from __future__ import annotations
+from ._base import _OpenNMSBase
+from typing import Any, Optional
 
 
-class AcksMixin:
+class AcksMixin(_OpenNMSBase):
     def get_acks(self, limit: int = 10, offset: int = 0, **filters):
         """List acknowledgements.
 
@@ -10,7 +13,7 @@ def get_acks(self, limit: int = 10, offset: int = 0, **filters):
             offset: Zero-based offset for pagination.
             **filters: Additional Hibernate query filters passed directly as query parameters (e.g. ``severity="MAJOR"``).
         """
-        params = {"limit": limit, "offset": offset}
+        params: dict[str, Any] = {"limit": limit, "offset": offset}
         params.update(filters)
         return self._get("acks", params=params)
 
@@ -24,8 +27,8 @@ def get_ack_count(self) -> int:
 
     # Write (form-encoded POST)
 
-    def create_ack(self, action: str, alarm_id: int = None,
-                   notification_id: int = None):
+    def create_ack(self, action: str, alarm_id: Optional[int] = None,
+                   notification_id: Optional[int] = None):
         """Create or modify an acknowledgement.
 
         Args:
@@ -34,7 +37,7 @@ def create_ack(self, action: str, alarm_id: int = None,
                 notification_id).
             notification_id: Target notification ID.
         """
-        data = {"action": action}
+        data: dict[str, Any] = {"action": action}
         if alarm_id is not None:
             data["alarmId"] = alarm_id
         if notification_id is not None:

opennms_api_wrapper/_alarm_history.py

@@ -1,8 +1,10 @@
 """Alarm History REST API – /rest/alarms/history."""
+from ._base import _OpenNMSBase
+from typing import Optional
 
 
-class AlarmHistoryMixin:
-    def get_alarm_history(self, at: int = None):
+class AlarmHistoryMixin(_OpenNMSBase):
+    def get_alarm_history(self, at: Optional[int] = None):
         """Return last known state of all active alarms.
 
         Args:
@@ -13,7 +15,7 @@ def get_alarm_history(self, at: int = None):
             params["at"] = at
         return self._get("alarms/history", params=params)
 
-    def get_alarm_history_at(self, alarm_id: int, at: int = None):
+    def get_alarm_history_at(self, alarm_id: int, at: Optional[int] = None):
         """Return final known state of *alarm_id* (optionally at a point in time).
 
         Args:

opennms_api_wrapper/_alarm_stats.py

@@ -1,7 +1,9 @@
 """Alarm Statistics REST API – /rest/stats/alarms."""
+from ._base import _OpenNMSBase
+from typing import Optional
 
 
-class AlarmStatsMixin:
+class AlarmStatsMixin(_OpenNMSBase):
     def get_alarm_stats(self, **filters):
         """Return alarm statistics.
 
@@ -11,7 +13,7 @@ def get_alarm_stats(self, **filters):
         """
         return self._get("stats/alarms", params=filters)
 
-    def get_alarm_stats_by_severity(self, severities: list = None):
+    def get_alarm_stats_by_severity(self, severities: Optional[list] = None):
         """Return alarm statistics grouped by severity.
 
         Args:

opennms_api_wrapper/_alarms.py

@@ -1,9 +1,12 @@
 """Alarms REST API – /rest/alarms and /api/v2/alarms."""
+from __future__ import annotations
+from ._base import _OpenNMSBase
+from typing import Any, Optional
 
 
-class AlarmsMixin:
+class AlarmsMixin(_OpenNMSBase):
     def get_alarms(self, limit: int = 10, offset: int = 0,
-                   order_by: str = None, order: str = None, **filters):
+                   order_by: Optional[str] = None, order: Optional[str] = None, **filters):
         """List alarms (v1).
 
         Args:
@@ -15,7 +18,7 @@ def get_alarms(self, limit: int = 10, offset: int = 0,
                 query parameters (e.g. ``severity="MAJOR"``). Pass
                 ``comparator`` to change the match type (eq/ilike/…).
         """
-        params = {"limit": limit, "offset": offset}
+        params: dict[str, Any] = {"limit": limit, "offset": offset}
         if order_by:
             params["orderBy"] = order_by
         if order:
@@ -33,13 +36,13 @@ def get_alarm_count(self) -> int:
 
     # Single-alarm actions (v1 PUT with query params)
 
-    def ack_alarm(self, alarm_id: int, ack_user: str = None):
+    def ack_alarm(self, alarm_id: int, ack_user: Optional[str] = None):
         """Acknowledge alarm *alarm_id*.
 
         Args:
             ack_user: Acknowledge on behalf of this user (requires admin role).
         """
-        params = {"ack": "true"}
+        params: dict[str, Any] = {"ack": "true"}
         if ack_user:
             params["ackUser"] = ack_user
         return self._put(f"alarms/{alarm_id}", params=params)
@@ -64,7 +67,7 @@ def bulk_ack_alarms(self, **filters):
         Args:
             **filters: Additional Hibernate query filters passed directly as query parameters (e.g. ``severity="MAJOR"``).
         """
-        params = {"ack": "true"}
+        params: dict[str, Any] = {"ack": "true"}
         params.update(filters)
         return self._put("alarms", params=params)
 
@@ -74,7 +77,7 @@ def bulk_unack_alarms(self, **filters):
         Args:
             **filters: Additional Hibernate query filters passed directly as query parameters (e.g. ``severity="MAJOR"``).
         """
-        params = {"ack": "false"}
+        params: dict[str, Any] = {"ack": "false"}
         params.update(filters)
         return self._put("alarms", params=params)
 
@@ -84,7 +87,7 @@ def bulk_clear_alarms(self, **filters):
         Args:
             **filters: Additional Hibernate query filters passed directly as query parameters (e.g. ``severity="MAJOR"``).
         """
-        params = {"clear": "true"}
+        params: dict[str, Any] = {"clear": "true"}
         params.update(filters)
         return self._put("alarms", params=params)
 
@@ -94,15 +97,15 @@ def bulk_escalate_alarms(self, **filters):
         Args:
             **filters: Additional Hibernate query filters passed directly as query parameters (e.g. ``severity="MAJOR"``).
         """
-        params = {"escalate": "true"}
+        params: dict[str, Any] = {"escalate": "true"}
         params.update(filters)
         return self._put("alarms", params=params)
 
     # v2 alarms (FIQL filtering)
 
-    def get_alarms_v2(self, fiql: str = None, limit: int = 10,
-                      offset: int = 0, order_by: str = None,
-                      order: str = None):
+    def get_alarms_v2(self, fiql: Optional[str] = None, limit: int = 10,
+                      offset: int = 0, order_by: Optional[str] = None,
+                      order: Optional[str] = None):
         """List alarms using the v2 API with optional FIQL filter string.
 
         Args:
@@ -116,7 +119,7 @@ def get_alarms_v2(self, fiql: str = None, limit: int = 10,
 
             client.get_alarms_v2(fiql="alarm.severity==MAJOR")
         """
-        params = {"limit": limit, "offset": offset}
+        params: dict[str, Any] = {"limit": limit, "offset": offset}
         if fiql:
             params["_s"] = fiql
         if order_by:

opennms_api_wrapper/_applications.py

@@ -1,8 +1,9 @@
 """Applications REST API v2 – /api/v2/applications."""
+from ._base import _OpenNMSBase
 from .types import Application
 
 
-class ApplicationsMixin:
+class ApplicationsMixin(_OpenNMSBase):
     def get_applications(self, limit: int = 10, offset: int = 0):
         """List all applications.
 

opennms_api_wrapper/_availability.py

@@ -1,7 +1,8 @@
 """Availability REST API – /rest/availability."""
+from ._base import _OpenNMSBase
 
 
-class AvailabilityMixin:
+class AvailabilityMixin(_OpenNMSBase):
     def get_availability(self):
         """Get availability summary for all categories."""
         return self._get("availability")

opennms_api_wrapper/_base.py

@@ -1,4 +1,6 @@
 """Base HTTP client for the OpenNMS REST API."""
+from __future__ import annotations
+from typing import Any, Optional
 import requests
 from requests.adapters import HTTPAdapter
 from requests.exceptions import HTTPError
@@ -106,14 +108,14 @@ def _parse(self, resp: requests.Response):
         except Exception:
             return resp.text
 
-    def _get(self, path: str, params: dict = None, v2: bool = False):
+    def _get(self, path: str, params: Optional[dict[str, Any]] = None, v2: bool = False):
         """Send a GET request and return the parsed response."""
         resp = self._session.get(self._url(path, v2), params=params,
                                  timeout=self._timeout)
         return self._parse(resp)
 
-    def _post(self, path: str, json_data=None, form_data: dict = None,
-              params: dict = None, v2: bool = False):
+    def _post(self, path: str, json_data=None, form_data: Optional[Any] = None,
+              params: Optional[dict[str, Any]] = None, v2: bool = False):
         """Send a POST request and return the parsed response.
 
         Sends form-encoded data when *form_data* is provided, otherwise JSON.
@@ -128,8 +130,8 @@ def _post(self, path: str, json_data=None, form_data: dict = None,
                                       timeout=self._timeout)
         return self._parse(resp)
 
-    def _put(self, path: str, json_data=None, form_data: dict = None,
-             params: dict = None, v2: bool = False):
+    def _put(self, path: str, json_data=None, form_data: Optional[Any] = None,
+             params: Optional[dict[str, Any]] = None, v2: bool = False):
         """Send a PUT request and return the parsed response.
 
         Sends form-encoded data when *form_data* is provided, otherwise JSON.
@@ -144,14 +146,14 @@ def _put(self, path: str, json_data=None, form_data: dict = None,
                                      timeout=self._timeout)
         return self._parse(resp)
 
-    def _delete(self, path: str, params: dict = None, json_data=None,
+    def _delete(self, path: str, params: Optional[dict[str, Any]] = None, json_data=None,
                 v2: bool = False):
         """Send a DELETE request and return the parsed response."""
         resp = self._session.delete(self._url(path, v2), params=params,
                                     json=json_data, timeout=self._timeout)
         return self._parse(resp)
 
-    def _patch(self, path: str, json_data=None, params: dict = None,
+    def _patch(self, path: str, json_data=None, params: Optional[dict[str, Any]] = None,
                v2: bool = False):
         """Send a PATCH request and return the parsed response."""
         resp = self._session.patch(self._url(path, v2), json=json_data,

opennms_api_wrapper/_business_services.py

@@ -1,8 +1,9 @@
 """Business Service Monitoring REST API v2 – /api/v2/business-services."""
+from ._base import _OpenNMSBase
 from .types import BusinessService, BsIpServiceEdge, BsReductionKeyEdge, BsChildEdge
 
 
-class BusinessServicesMixin:
+class BusinessServicesMixin(_OpenNMSBase):
     def get_business_services(self):
         """List all business services."""
         return self._get("business-services", v2=True)

opennms_api_wrapper/_categories.py

@@ -1,8 +1,9 @@
 """Categories REST API – /rest/categories."""
+from ._base import _OpenNMSBase
 from .types import Category
 
 
-class CategoriesMixin:
+class CategoriesMixin(_OpenNMSBase):
     # ==================================================================
     # Categories CRUD
     # ==================================================================

opennms_api_wrapper/_classifications.py

@@ -1,8 +1,10 @@
 """Flow Classification REST API – /rest/classifications."""
+from ._base import _OpenNMSBase
+from typing import Optional
 from .types import ClassificationRule, ClassificationGroup, ClassifyRequest
 
 
-class ClassificationsMixin:
+class ClassificationsMixin(_OpenNMSBase):
     # ------------------------------------------------------------------
     # Rules
     # ------------------------------------------------------------------
@@ -40,7 +42,7 @@ def update_classification_rule(self, rule_id: int, rule: ClassificationRule):
         """
         return self._put(f"classifications/{rule_id}", json_data=rule)
 
-    def delete_classification_rules(self, group_id: int = None):
+    def delete_classification_rules(self, group_id: Optional[int] = None):
         """Delete classification rules, optionally filtered by group.
 
         Args:

opennms_api_wrapper/_config_mgmt.py

@@ -1,7 +1,8 @@
 """Configuration Management REST API – /rest/cm."""
+from ._base import _OpenNMSBase
 
 
-class ConfigMgmtMixin:
+class ConfigMgmtMixin(_OpenNMSBase):
     def get_config_names(self):
         """List all configuration names."""
         return self._get("cm")

opennms_api_wrapper/_device_config.py

@@ -1,13 +1,15 @@
 """Device Configuration REST API – /rest/device-config."""
+from ._base import _OpenNMSBase
+from typing import Optional
 
 
-class DeviceConfigMixin:
+class DeviceConfigMixin(_OpenNMSBase):
     def get_device_configs(self, limit: int = 10, offset: int = 0,
-                           order_by: str = None, order: str = None,
-                           device_name: str = None, ip_address: str = None,
-                           config_type: str = None,
-                           created_after: int = None,
-                           created_before: int = None):
+                           order_by: Optional[str] = None, order: Optional[str] = None,
+                           device_name: Optional[str] = None, ip_address: Optional[str] = None,
+                           config_type: Optional[str] = None,
+                           created_after: Optional[int] = None,
+                           created_before: Optional[int] = None):
         """List all device configurations (sorted by lastUpdated by default).
 
         Args:
@@ -47,8 +49,8 @@ def get_device_config_by_interface(self, interface_id: int):
         return self._get(f"device-config/interface/{interface_id}")
 
     def get_latest_device_configs(self, limit: int = 10, offset: int = 0,
-                                  order_by: str = None, order: str = None,
-                                  search: str = None, status: str = None):
+                                  order_by: Optional[str] = None, order: Optional[str] = None,
+                                  search: Optional[str] = None, status: Optional[str] = None):
         """Return the latest config for all devices.
 
         Args:

opennms_api_wrapper/_discovery.py

@@ -1,7 +1,8 @@
 """Discovery REST API v2 – /api/v2/discovery."""
+from ._base import _OpenNMSBase
 
 
-class DiscoveryMixin:
+class DiscoveryMixin(_OpenNMSBase):
     def discover(self, config: dict):
         """Submit a one-time discovery scan configuration (v2).
 

opennms_api_wrapper/_email_nbi.py

@@ -1,8 +1,9 @@
 """Email NBI Configuration REST API – /rest/config/email-nbi."""
+from ._base import _OpenNMSBase
 from .types import EmailNbiDestination, EmailNbiConfig
 
 
-class EmailNbiMixin:
+class EmailNbiMixin(_OpenNMSBase):
     _EMAIL_NBI = "config/email-nbi"
 
     def get_email_nbi_config(self):

opennms_api_wrapper/_eventconf.py

@@ -1,13 +1,15 @@
 """Event Configuration REST API v2 – /api/v2/eventconf."""
+from ._base import _OpenNMSBase
+from typing import Optional
 from .types import EventConfEvent
 
 
-class EventConfMixin:
+class EventConfMixin(_OpenNMSBase):
     # ------------------------------------------------------------------
     # Filter
     # ------------------------------------------------------------------
 
-    def get_eventconf_filter(self, uei: str = None, vendor: str = None,
+    def get_eventconf_filter(self, uei: Optional[str] = None, vendor: Optional[str] = None,
                              **kwargs):
         """Get event configuration with optional filtering.
 
@@ -24,7 +26,7 @@ def get_eventconf_filter(self, uei: str = None, vendor: str = None,
         return self._get("eventconf/filter", params=params,
                          v2=True)
 
-    def get_eventconf_filter_sources(self, filter: str = None,
+    def get_eventconf_filter_sources(self, filter: Optional[str] = None,
                                      **kwargs):
         """Get event configuration sources.