diff --git a/docs/superpowers/plans/2026-04-29-mypy-fixes.md b/docs/superpowers/plans/2026-04-29-mypy-fixes.md new file mode 100644 index 0000000..dc6312c --- /dev/null +++ b/docs/superpowers/plans/2026-04-29-mypy-fixes.md @@ -0,0 +1,439 @@ +# mypy Fixes Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Fix all 427 mypy errors so `mypy opennms_api_wrapper/` exits 0 while keeping `pytest tests/ -v` green. + +**Architecture:** Four mechanical change types applied across ~50 files. The biggest fix (321 `attr-defined` errors) is solved by making each mixin inherit from `_OpenNMSBase` — Python's MRO handles the resulting diamond inheritance in `client.py` correctly. The other three fixes are type-annotation cleanups. + +**Tech Stack:** Python 3.8+, mypy, `typing.Optional`, `typing.Any`, `types-requests` + +--- + +## Error categories + +| Code | Count | Fix | +|------|-------|-----| +| `attr-defined` | 321 | `class FooMixin(_OpenNMSBase):` + `from ._base import _OpenNMSBase` | +| `no_implicit_optional` | 89 | `param: str = None` → `param: Optional[str] = None` | +| `str↔int assignment` | 14 | `params: dict[str, Any] = {...}` explicit annotation | +| `import-untyped` | 3 | Add `types-requests` to dev deps | + +--- + +## Shared pattern reference + +Every mixin that triggers `attr-defined` needs these two changes — the rest of the file is untouched: + +```python +# 1. Add after the module docstring (before any existing imports): +from ._base import _OpenNMSBase + +# 2. Change the class definition line: +class FooMixin(_OpenNMSBase): # was: class FooMixin: +``` + +For `no_implicit_optional`, add/extend the typing import and annotate the parameter: + +```python +from typing import Any, Optional # add to existing import or create new one + +def method(self, param: str = None, ...): # before +def method(self, param: Optional[str] = None, ...): # after +``` + +For `str↔int` dict assignment, add an explicit annotation at the dict literal: + +```python +params: dict[str, Any] = {"limit": limit, "offset": offset} # was: params = {...} +``` + +--- + +## Task 1: pyproject.toml + _base.py + +**Files:** +- Modify: `pyproject.toml` +- Modify: `opennms_api_wrapper/_base.py` + +- [ ] **Add `types-requests` to dev deps** + + In `pyproject.toml`, change: + ```toml + dev = ["pytest", "pytest-cov", "responses", "mkdocs-material", "mkdocstrings[python]", "ruff", "mypy"] + ``` + to: + ```toml + dev = ["pytest", "pytest-cov", "responses", "mkdocs-material", "mkdocstrings[python]", "ruff", "mypy", "types-requests"] + ``` + +- [ ] **Install the new dep** + + ```bash + pip install -e ".[dev]" -q + ``` + +- [ ] **Fix `_base.py` — add Optional import and annotate params** + + Add `from typing import Any, Optional` after the existing imports (after line 10, before the `from ._exceptions` block — or just append it): + + ```python + from typing import Any, Optional + ``` + + Then fix the three helper signatures that use `params: dict = None` (lines ~109, 147, 154) and the `timeout` param (line 17 `timeout: int = 30` is fine — the `None` case is `timeout: Optional[int] = 30` only if you need to pass None; check actual errors): + + Run first to see the exact lines: + ```bash + mypy opennms_api_wrapper/_base.py 2>&1 | grep "error:" + ``` + + For each `params: dict = None`, change to `params: Optional[dict[str, Any]] = None`. + +- [ ] **Verify _base.py is clean** + + ```bash + mypy opennms_api_wrapper/_base.py + ``` + Expected: `Success: no issues found in 1 source file` + +- [ ] **Commit** + + ```bash + git add pyproject.toml opennms_api_wrapper/_base.py + git commit -m "fix(types): add types-requests dev dep and fix _base.py Optional params" + ``` + +--- + +## Task 2: Mixin batch A — alarms, acks, events, nodes (all 3 fix types) + +These files need the mixin inheritance fix **plus** `Optional` annotations **plus** `dict[str, Any]` dict annotation. + +**Files:** `_alarms.py`, `_acks.py`, `_events.py`, `_nodes.py`, `_outages.py`, `_notifications.py`, `_ifservices.py` + +- [ ] **Check exact errors for each file** + + ```bash + mypy opennms_api_wrapper/_alarms.py opennms_api_wrapper/_acks.py \ + opennms_api_wrapper/_events.py opennms_api_wrapper/_nodes.py \ + opennms_api_wrapper/_outages.py opennms_api_wrapper/_notifications.py \ + opennms_api_wrapper/_ifservices.py 2>&1 | grep "error:" + ``` + +- [ ] **Fix `_alarms.py`** + + Add at top: + ```python + from typing import Any, Optional + from ._base import _OpenNMSBase + ``` + + Change class line: + ```python + class AlarmsMixin(_OpenNMSBase): + ``` + + For every `params = {"limit": ..., "offset": ...}` dict literal that later gets string keys assigned, add explicit annotation: + ```python + params: dict[str, Any] = {"limit": limit, "offset": offset} + ``` + + For `order_by: str = None` and `order: str = None` parameters, change to `Optional[str] = None`. + +- [ ] **Fix `_acks.py`** + + Add at top: + ```python + from typing import Any, Optional + from ._base import _OpenNMSBase + ``` + Change class: `class AcksMixin(_OpenNMSBase):` + + In `create_ack`, `alarm_id: int = None` → `alarm_id: Optional[int] = None`, same for `notification_id`. + + `data = {"action": action}` → `data: dict[str, Any] = {"action": action}` + +- [ ] **Fix `_events.py`** + + Add at top: + ```python + from typing import Any, Optional + from ._base import _OpenNMSBase + ``` + Change class: `class EventsMixin(_OpenNMSBase):` + + Find the `params = {"limit": ..., "offset": ...}` dict that later gets string keys and annotate: `params: dict[str, Any] = {...}`. + + Fix any `param: str = None` signatures with `Optional[str]`. + +- [ ] **Fix `_nodes.py`** + + `_nodes.py` already imports from `.types` — add the new imports on separate lines before it: + ```python + from typing import Any, Optional + from ._base import _OpenNMSBase + from .types import Node, NodeIpInterface, NodeSnmpInterface, NodeAssetRecord, HardwareEntity, Category + ``` + Change class: `class NodesMixin(_OpenNMSBase):` + + Find `params = {"limit": ..., "offset": ...}` dicts and annotate as `dict[str, Any]`. + Fix `Optional` parameters as indicated by mypy. + +- [ ] **Fix `_outages.py`, `_notifications.py`, `_ifservices.py`** + + Same pattern: add typing + `_OpenNMSBase` imports, update class line, annotate mixed-type dicts as `dict[str, Any]`, fix `Optional` params. + +- [ ] **Verify batch A** + + ```bash + mypy opennms_api_wrapper/_alarms.py opennms_api_wrapper/_acks.py \ + opennms_api_wrapper/_events.py opennms_api_wrapper/_nodes.py \ + opennms_api_wrapper/_outages.py opennms_api_wrapper/_notifications.py \ + opennms_api_wrapper/_ifservices.py + ``` + Expected: `Success: no issues found in 7 source files` + +- [ ] **Run tests** + + ```bash + pytest tests/test_alarms.py tests/test_acks.py tests/test_events.py \ + tests/test_nodes.py tests/test_outages.py tests/test_notifications.py \ + tests/test_ifservices.py -v + ``` + Expected: all PASS + +- [ ] **Commit** + + ```bash + git add opennms_api_wrapper/_alarms.py opennms_api_wrapper/_acks.py \ + opennms_api_wrapper/_events.py opennms_api_wrapper/_nodes.py \ + opennms_api_wrapper/_outages.py opennms_api_wrapper/_notifications.py \ + opennms_api_wrapper/_ifservices.py + git commit -m "fix(types): mixin inheritance + Optional + dict[str,Any] for alarms/acks/events/nodes batch" + ``` + +--- + +## Task 3: Mixin batch B — flows, device_config, resources, graphs, perspective_poller (mixin + Optional) + +**Files:** `_flows.py`, `_device_config.py`, `_resources.py`, `_graphs.py`, `_perspective_poller.py` + +These have high `no_implicit_optional` counts (24, 11, 4, 4, 4) on top of the mixin fix. + +- [ ] **Check exact errors** + + ```bash + mypy opennms_api_wrapper/_flows.py opennms_api_wrapper/_device_config.py \ + opennms_api_wrapper/_resources.py opennms_api_wrapper/_graphs.py \ + opennms_api_wrapper/_perspective_poller.py 2>&1 | grep "error:" + ``` + +- [ ] **Fix each file** + + For each file, add: + ```python + from typing import Any, Optional + from ._base import _OpenNMSBase + ``` + Update the class line: `class FlowsMixin(_OpenNMSBase):` etc. + + Then for every parameter reported by mypy as `"default has type None, parameter has type X"`, change `param: X = None` to `param: Optional[X] = None`. + + `_flows.py` has 24 optional errors — `if_index: int = None` → `Optional[int]` and `exporter_node: str = None` → `Optional[str]`, repeated across many methods. Find-and-replace within the file. + + `_resources.py` has `nodes: list[Any] = None` → `Optional[list[Any]]` etc. + +- [ ] **Verify batch B** + + ```bash + mypy opennms_api_wrapper/_flows.py opennms_api_wrapper/_device_config.py \ + opennms_api_wrapper/_resources.py opennms_api_wrapper/_graphs.py \ + opennms_api_wrapper/_perspective_poller.py + ``` + Expected: `Success: no issues found in 5 source files` + +- [ ] **Run tests** + + ```bash + pytest tests/test_flows.py tests/test_device_config.py tests/test_resources.py \ + tests/test_graphs.py tests/test_perspective_poller.py -v + ``` + Expected: all PASS + +- [ ] **Commit** + + ```bash + git add opennms_api_wrapper/_flows.py opennms_api_wrapper/_device_config.py \ + opennms_api_wrapper/_resources.py opennms_api_wrapper/_graphs.py \ + opennms_api_wrapper/_perspective_poller.py + git commit -m "fix(types): mixin inheritance + Optional for flows/device_config/resources/graphs batch" + ``` + +--- + +## Task 4: Mixin batch C — situations, eventconf, alarm_history, snmp/ip interfaces, feedback, measurements, health, classifications, alarm_stats (mixin + Optional) + +**Files:** `_situations.py`, `_eventconf.py`, `_alarm_history.py`, `_snmpinterfaces_v2.py`, `_snmp_config.py`, `_situation_feedback.py`, `_measurements.py`, `_ipinterfaces_v2.py`, `_ifservices.py` (already done in Task 2), `_health.py`, `_classifications.py`, `_alarm_stats.py` + +- [ ] **Check exact errors** + + ```bash + mypy opennms_api_wrapper/_situations.py opennms_api_wrapper/_eventconf.py \ + opennms_api_wrapper/_alarm_history.py opennms_api_wrapper/_snmpinterfaces_v2.py \ + opennms_api_wrapper/_snmp_config.py opennms_api_wrapper/_situation_feedback.py \ + opennms_api_wrapper/_measurements.py opennms_api_wrapper/_ipinterfaces_v2.py \ + opennms_api_wrapper/_health.py opennms_api_wrapper/_classifications.py \ + opennms_api_wrapper/_alarm_stats.py 2>&1 | grep "error:" + ``` + +- [ ] **Fix each file** using the shared pattern + + Add to each file: + ```python + from typing import Any, Optional + from ._base import _OpenNMSBase + ``` + Update class line to inherit from `_OpenNMSBase`. + Fix `param: str = None` → `param: Optional[str] = None` as flagged by mypy. + + `_situations.py` has 3 optional params: `description`, `diagnostic_text`, `feedback` — all `str = None` → `Optional[str] = None`. + `_snmpinterfaces_v2.py` has `fiql: str = None` → `Optional[str]`. + `_situation_feedback.py` has `prefix: str = None` → `Optional[str]`. + +- [ ] **Verify batch C** + + ```bash + mypy opennms_api_wrapper/_situations.py opennms_api_wrapper/_eventconf.py \ + opennms_api_wrapper/_alarm_history.py opennms_api_wrapper/_snmpinterfaces_v2.py \ + opennms_api_wrapper/_snmp_config.py opennms_api_wrapper/_situation_feedback.py \ + opennms_api_wrapper/_measurements.py opennms_api_wrapper/_ipinterfaces_v2.py \ + opennms_api_wrapper/_health.py opennms_api_wrapper/_classifications.py \ + opennms_api_wrapper/_alarm_stats.py + ``` + Expected: `Success: no issues found in 11 source files` + +- [ ] **Run tests** + + ```bash + pytest tests/test_situations.py tests/test_alarm_history.py \ + tests/test_alarm_stats.py tests/test_classifications.py \ + tests/test_health.py -v + ``` + Expected: all PASS + +- [ ] **Commit** + + ```bash + git add opennms_api_wrapper/_situations.py opennms_api_wrapper/_eventconf.py \ + opennms_api_wrapper/_alarm_history.py opennms_api_wrapper/_snmpinterfaces_v2.py \ + opennms_api_wrapper/_snmp_config.py opennms_api_wrapper/_situation_feedback.py \ + opennms_api_wrapper/_measurements.py opennms_api_wrapper/_ipinterfaces_v2.py \ + opennms_api_wrapper/_health.py opennms_api_wrapper/_classifications.py \ + opennms_api_wrapper/_alarm_stats.py + git commit -m "fix(types): mixin inheritance + Optional for situations/eventconf/snmp/measurements batch" + ``` + +--- + +## Task 5: Mixin batch D — mixin-only files (no Optional errors, just inheritance) + +The remaining 31 mixin files need only the inheritance fix (no Optional or dict annotation issues). + +**Files:** +`_applications.py`, `_availability.py`, `_business_services.py`, `_categories.py`, +`_config_mgmt.py`, `_device_config.py` (if not done in batch B), `_discovery.py`, +`_email_nbi.py`, `_enlinkd.py`, `_foreign_sources.py`, `_foreign_sources_config.py`, +`_groups.py`, `_heatmap.py`, `_info.py` (check — may not have attr-defined errors), +`_javamail_config.py`, `_ksc_reports.py`, `_maps.py`, `_metadata.py`, +`_minions.py`, `_monitoring_locations.py`, `_monitoring_systems.py`, +`_provisiond.py`, `_requisition_names.py`, `_requisitions.py`, +`_resources.py` (if not done in batch B), `_sched_outages.py`, `_scv.py`, +`_situation_feedback.py` (if not done in batch C), `_situations.py` (if not done), +`_snmp_metadata.py`, `_snmptrap_nbi.py`, `_syslog_nbi.py`, +`_user_defined_links.py`, `_users.py`, `_whoami.py` + +> Note: Cross-check against the attr-defined list. Files not in that list (e.g. `_pagination.py`, `_exceptions.py`) need no changes. + +- [ ] **Verify which files still need the fix** + + ```bash + source .venv/bin/activate + mypy opennms_api_wrapper/ 2>&1 | grep "attr-defined" | sed 's|opennms_api_wrapper/||' | sed 's/:[0-9]*.*//' | sort -u + ``` + +- [ ] **Apply the mixin inheritance pattern to each remaining file** + + For files without any existing imports: + ```python + """Original module docstring.""" + from ._base import _OpenNMSBase # ADD THIS LINE + + + class FooMixin(_OpenNMSBase): # ADD _OpenNMSBase + ``` + + For files that already import from `.types`: + ```python + """Original module docstring.""" + from ._base import _OpenNMSBase # ADD BEFORE existing imports + from .types import SomeType # existing — keep as-is + + + class FooMixin(_OpenNMSBase): # ADD _OpenNMSBase + ``` + +- [ ] **Verify all batch D files** + + ```bash + mypy opennms_api_wrapper/ 2>&1 | grep "attr-defined" | wc -l + ``` + Expected: `0` + +- [ ] **Run full test suite** + + ```bash + pytest tests/ -v + ``` + Expected: 490 passed + +- [ ] **Commit** + + ```bash + git add opennms_api_wrapper/ + git commit -m "fix(types): mixin inheritance for remaining batch D files" + ``` + +--- + +## Task 6: Final verification and PR + +- [ ] **Run full mypy check** + + ```bash + mypy opennms_api_wrapper/ + ``` + Expected: `Success: no issues found in N source files` + +- [ ] **Run full test suite** + + ```bash + pytest tests/ -v + ``` + Expected: 490 passed, 0 failed + +- [ ] **Run ruff** + + ```bash + ruff check opennms_api_wrapper/ + ``` + Expected: `All checks passed!` + +- [ ] **Open PR** + + ```bash + git push -u origin fix/mypy-clean + gh pr create \ + --title "fix(types): resolve all mypy errors across mixin package" \ + --body "Fixes 427 pre-existing mypy errors across 50 files. Four change types: mixin inheritance (_OpenNMSBase base), Optional annotations, dict[str,Any] dict literals, types-requests dev dep. All 490 tests pass." + ``` diff --git a/opennms_api_wrapper/_acks.py b/opennms_api_wrapper/_acks.py index 169f680..17cd05e 100644 --- a/opennms_api_wrapper/_acks.py +++ b/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: diff --git a/opennms_api_wrapper/_alarm_history.py b/opennms_api_wrapper/_alarm_history.py index b85750c..b0a953c 100644 --- a/opennms_api_wrapper/_alarm_history.py +++ b/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: diff --git a/opennms_api_wrapper/_alarm_stats.py b/opennms_api_wrapper/_alarm_stats.py index b68e8e5..1f79c12 100644 --- a/opennms_api_wrapper/_alarm_stats.py +++ b/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: diff --git a/opennms_api_wrapper/_alarms.py b/opennms_api_wrapper/_alarms.py index f2ba851..add9413 100644 --- a/opennms_api_wrapper/_alarms.py +++ b/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: diff --git a/opennms_api_wrapper/_applications.py b/opennms_api_wrapper/_applications.py index df33bef..5b7cf4d 100644 --- a/opennms_api_wrapper/_applications.py +++ b/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. diff --git a/opennms_api_wrapper/_availability.py b/opennms_api_wrapper/_availability.py index 46d704d..362ed62 100644 --- a/opennms_api_wrapper/_availability.py +++ b/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") diff --git a/opennms_api_wrapper/_base.py b/opennms_api_wrapper/_base.py index 526cecd..137ef7f 100644 --- a/opennms_api_wrapper/_base.py +++ b/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, diff --git a/opennms_api_wrapper/_business_services.py b/opennms_api_wrapper/_business_services.py index f516aa6..da02edd 100644 --- a/opennms_api_wrapper/_business_services.py +++ b/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) diff --git a/opennms_api_wrapper/_categories.py b/opennms_api_wrapper/_categories.py index 89a6d51..b50e56f 100644 --- a/opennms_api_wrapper/_categories.py +++ b/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 # ================================================================== diff --git a/opennms_api_wrapper/_classifications.py b/opennms_api_wrapper/_classifications.py index 03467a9..31dc379 100644 --- a/opennms_api_wrapper/_classifications.py +++ b/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: diff --git a/opennms_api_wrapper/_config_mgmt.py b/opennms_api_wrapper/_config_mgmt.py index 6c13d9e..7078602 100644 --- a/opennms_api_wrapper/_config_mgmt.py +++ b/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") diff --git a/opennms_api_wrapper/_device_config.py b/opennms_api_wrapper/_device_config.py index 39c3027..c85b6d1 100644 --- a/opennms_api_wrapper/_device_config.py +++ b/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: diff --git a/opennms_api_wrapper/_discovery.py b/opennms_api_wrapper/_discovery.py index 63589d2..4fe17f3 100644 --- a/opennms_api_wrapper/_discovery.py +++ b/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). diff --git a/opennms_api_wrapper/_email_nbi.py b/opennms_api_wrapper/_email_nbi.py index 637b12b..e9fdb21 100644 --- a/opennms_api_wrapper/_email_nbi.py +++ b/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): diff --git a/opennms_api_wrapper/_eventconf.py b/opennms_api_wrapper/_eventconf.py index 59c87e1..10bbc3c 100644 --- a/opennms_api_wrapper/_eventconf.py +++ b/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. diff --git a/opennms_api_wrapper/_events.py b/opennms_api_wrapper/_events.py index 6d781aa..ef32433 100644 --- a/opennms_api_wrapper/_events.py +++ b/opennms_api_wrapper/_events.py @@ -1,10 +1,13 @@ """Events REST API – /rest/events.""" +from __future__ import annotations +from ._base import _OpenNMSBase +from typing import Any, Optional from .types import Event -class EventsMixin: +class EventsMixin(_OpenNMSBase): def get_events(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 events. Args: @@ -14,7 +17,7 @@ def get_events(self, limit: int = 10, offset: int = 0, order: Sort direction: ``"asc"`` or ``"desc"``. **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} if order_by: params["orderBy"] = order_by if order: @@ -69,7 +72,7 @@ def bulk_ack_events(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("events", params=params) @@ -79,6 +82,6 @@ def bulk_unack_events(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("events", params=params) diff --git a/opennms_api_wrapper/_flows.py b/opennms_api_wrapper/_flows.py index 1deb987..a52a517 100644 --- a/opennms_api_wrapper/_flows.py +++ b/opennms_api_wrapper/_flows.py @@ -1,7 +1,9 @@ """Flow REST API – /rest/flows (read-only).""" +from ._base import _OpenNMSBase +from typing import Optional -class FlowsMixin: +class FlowsMixin(_OpenNMSBase): # ------------------------------------------------------------------ # Overview # ------------------------------------------------------------------ @@ -31,8 +33,8 @@ def get_flow_exporter(self, node_criteria: str): # ------------------------------------------------------------------ def get_flow_applications(self, top_n: int = 10, start: int = -14400000, - end: int = 0, if_index: int = None, - exporter_node: str = None, + end: int = 0, if_index: Optional[int] = None, + exporter_node: Optional[str] = None, include_other: bool = False): """Return traffic stats for the top *top_n* applications. @@ -53,8 +55,8 @@ def get_flow_applications(self, top_n: int = 10, start: int = -14400000, return self._get("flows/applications", params=params) def get_flow_applications_enumerate(self, start: int = -14400000, - end: int = 0, if_index: int = None, - exporter_node: str = None, + end: int = 0, if_index: Optional[int] = None, + exporter_node: Optional[str] = None, limit: int = 10): """List application names that have flows in the given time range. @@ -73,8 +75,8 @@ def get_flow_applications_enumerate(self, start: int = -14400000, def get_flow_applications_series(self, top_n: int = 10, start: int = -14400000, end: int = 0, step: int = 300000, - if_index: int = None, - exporter_node: str = None, + if_index: Optional[int] = None, + exporter_node: Optional[str] = None, include_other: bool = False): """Return time-series data for the top *top_n* applications. @@ -98,8 +100,8 @@ def get_flow_applications_series(self, top_n: int = 10, # ------------------------------------------------------------------ def get_flow_conversations(self, top_n: int = 10, start: int = -14400000, - end: int = 0, if_index: int = None, - exporter_node: str = None, + end: int = 0, if_index: Optional[int] = None, + exporter_node: Optional[str] = None, include_other: bool = False): """Return traffic stats for the top *top_n* conversations. @@ -118,8 +120,8 @@ def get_flow_conversations(self, top_n: int = 10, start: int = -14400000, return self._get("flows/conversations", params=params) def get_flow_conversations_enumerate(self, start: int = -14400000, - end: int = 0, if_index: int = None, - exporter_node: str = None, + end: int = 0, if_index: Optional[int] = None, + exporter_node: Optional[str] = None, limit: int = 10): """List conversations that have flows in the given time range. @@ -138,8 +140,8 @@ def get_flow_conversations_enumerate(self, start: int = -14400000, def get_flow_conversations_series(self, top_n: int = 10, start: int = -14400000, end: int = 0, step: int = 300000, - if_index: int = None, - exporter_node: str = None, + if_index: Optional[int] = None, + exporter_node: Optional[str] = None, include_other: bool = False): """Return time-series data for the top *top_n* conversations. @@ -163,8 +165,8 @@ def get_flow_conversations_series(self, top_n: int = 10, # ------------------------------------------------------------------ def get_flow_hosts(self, top_n: int = 10, start: int = -14400000, - end: int = 0, if_index: int = None, - exporter_node: str = None, + end: int = 0, if_index: Optional[int] = None, + exporter_node: Optional[str] = None, include_other: bool = False): """Return traffic stats for the top *top_n* hosts. @@ -183,8 +185,8 @@ def get_flow_hosts(self, top_n: int = 10, start: int = -14400000, return self._get("flows/hosts", params=params) def get_flow_hosts_enumerate(self, start: int = -14400000, end: int = 0, - if_index: int = None, - exporter_node: str = None, limit: int = 10): + if_index: Optional[int] = None, + exporter_node: Optional[str] = None, limit: int = 10): """List hosts that have flows in the given time range. Args: @@ -201,7 +203,7 @@ def get_flow_hosts_enumerate(self, start: int = -14400000, end: int = 0, def get_flow_hosts_series(self, top_n: int = 10, start: int = -14400000, end: int = 0, step: int = 300000, - if_index: int = None, exporter_node: str = None, + if_index: Optional[int] = None, exporter_node: Optional[str] = None, include_other: bool = False): """Return time-series data for the top *top_n* hosts. @@ -225,8 +227,8 @@ def get_flow_hosts_series(self, top_n: int = 10, start: int = -14400000, # ------------------------------------------------------------------ def get_flow_dscp(self, top_n: int = 10, start: int = -14400000, - end: int = 0, if_index: int = None, - exporter_node: str = None, + end: int = 0, if_index: Optional[int] = None, + exporter_node: Optional[str] = None, include_other: bool = False): """Return traffic stats for the top *top_n* DSCP values. @@ -244,8 +246,8 @@ def get_flow_dscp(self, top_n: int = 10, start: int = -14400000, return self._get("flows/dscp", params=params) def get_flow_dscp_enumerate(self, start: int = -14400000, end: int = 0, - if_index: int = None, - exporter_node: str = None, + if_index: Optional[int] = None, + exporter_node: Optional[str] = None, limit: int = 10): """List DSCP values that have flows in the given time range. @@ -262,8 +264,8 @@ def get_flow_dscp_enumerate(self, start: int = -14400000, end: int = 0, def get_flow_dscp_series(self, top_n: int = 10, start: int = -14400000, end: int = 0, step: int = 300000, - if_index: int = None, - exporter_node: str = None, + if_index: Optional[int] = None, + exporter_node: Optional[str] = None, include_other: bool = False): """Return time-series data for the top *top_n* DSCP values. diff --git a/opennms_api_wrapper/_foreign_sources.py b/opennms_api_wrapper/_foreign_sources.py index 94bb16a..9d761a2 100644 --- a/opennms_api_wrapper/_foreign_sources.py +++ b/opennms_api_wrapper/_foreign_sources.py @@ -1,8 +1,9 @@ """Foreign Sources REST API – /rest/foreignSources.""" +from ._base import _OpenNMSBase from .types import ForeignSource, ForeignSourceDetector, ForeignSourcePolicy -class ForeignSourcesMixin: +class ForeignSourcesMixin(_OpenNMSBase): # ================================================================== # Foreign Sources # ================================================================== diff --git a/opennms_api_wrapper/_foreign_sources_config.py b/opennms_api_wrapper/_foreign_sources_config.py index bd0ea22..30d7f58 100644 --- a/opennms_api_wrapper/_foreign_sources_config.py +++ b/opennms_api_wrapper/_foreign_sources_config.py @@ -1,7 +1,8 @@ """Foreign Sources Configuration REST API – /rest/foreignSourcesConfig.""" +from ._base import _OpenNMSBase -class ForeignSourcesConfigMixin: +class ForeignSourcesConfigMixin(_OpenNMSBase): def get_foreign_source_config_policies(self): """List available provisioning policies.""" return self._get("foreignSourcesConfig/policies") diff --git a/opennms_api_wrapper/_graphs.py b/opennms_api_wrapper/_graphs.py index 5fb4e94..105f541 100644 --- a/opennms_api_wrapper/_graphs.py +++ b/opennms_api_wrapper/_graphs.py @@ -1,7 +1,9 @@ """Graph / Topology REST API – /rest/graphs.""" +from ._base import _OpenNMSBase +from typing import Optional -class GraphsMixin: +class GraphsMixin(_OpenNMSBase): def get_graph_containers(self): """List all registered graph containers and their metadata.""" return self._get("graphs") @@ -16,7 +18,7 @@ def get_graph(self, container_id: str, namespace: str): def get_graph_view(self, container_id: str, namespace: str, semantic_zoom_level: int = 1, - vertices_in_focus: list = None): + vertices_in_focus: Optional[list] = None): """Get a focused graph view via POST. Args: @@ -44,8 +46,8 @@ def get_graph_search_suggestions(self, namespace: str, search_term: str): params={"s": search_term}, ) - def get_graph_search_results(self, namespace: str, provider_id: str = None, - criteria: str = None, context: str = None): + def get_graph_search_results(self, namespace: str, provider_id: Optional[str] = None, + criteria: Optional[str] = None, context: Optional[str] = None): """Return search results for graph elements in *namespace*. Args: diff --git a/opennms_api_wrapper/_groups.py b/opennms_api_wrapper/_groups.py index 1cea2f0..c7095b1 100644 --- a/opennms_api_wrapper/_groups.py +++ b/opennms_api_wrapper/_groups.py @@ -1,8 +1,9 @@ """Groups REST API – /rest/groups.""" +from ._base import _OpenNMSBase from .types import Group -class GroupsMixin: +class GroupsMixin(_OpenNMSBase): # ================================================================== # Groups # ================================================================== diff --git a/opennms_api_wrapper/_health.py b/opennms_api_wrapper/_health.py index 2c15514..9c05fac 100644 --- a/opennms_api_wrapper/_health.py +++ b/opennms_api_wrapper/_health.py @@ -1,8 +1,10 @@ """Health REST API – /rest/health.""" +from ._base import _OpenNMSBase +from typing import Optional -class HealthMixin: - def get_health(self, tag: str = None): +class HealthMixin(_OpenNMSBase): + def get_health(self, tag: Optional[str] = None): """Get the health status of the OpenNMS instance. Args: diff --git a/opennms_api_wrapper/_heatmap.py b/opennms_api_wrapper/_heatmap.py index ee3f97d..85ef5e7 100644 --- a/opennms_api_wrapper/_heatmap.py +++ b/opennms_api_wrapper/_heatmap.py @@ -1,7 +1,8 @@ """Heatmap REST API – /rest/heatmap (read-only).""" +from ._base import _OpenNMSBase -class HeatmapMixin: +class HeatmapMixin(_OpenNMSBase): # ================================================================== # Outage-based heatmap # ================================================================== diff --git a/opennms_api_wrapper/_ifservices.py b/opennms_api_wrapper/_ifservices.py index 6e164ce..d78c084 100644 --- a/opennms_api_wrapper/_ifservices.py +++ b/opennms_api_wrapper/_ifservices.py @@ -1,7 +1,10 @@ """Monitored Services (ifservices) REST API – /rest/ifservices + v2.""" +from __future__ import annotations +from ._base import _OpenNMSBase +from typing import Any, Optional -class IfServicesMixin: +class IfServicesMixin(_OpenNMSBase): def get_ifservices(self, **kwargs): """List monitored services with optional query parameters. @@ -20,7 +23,7 @@ def update_ifservices(self, **kwargs): """ return self._put("ifservices", form_data=kwargs) - def get_ifservices_v2(self, fiql: str = None, limit: int = 10, + def get_ifservices_v2(self, fiql: Optional[str] = None, limit: int = 10, offset: int = 0): """List monitored services via the v2 API with FIQL filtering. @@ -29,7 +32,7 @@ def get_ifservices_v2(self, fiql: str = None, limit: int = 10, limit: Maximum number of results. offset: Number of results to skip. """ - params = {"limit": limit, "offset": offset} + params: dict[str, Any] = {"limit": limit, "offset": offset} if fiql: params["_s"] = fiql return self._get("ifservices", params=params, v2=True) diff --git a/opennms_api_wrapper/_ipinterfaces_v2.py b/opennms_api_wrapper/_ipinterfaces_v2.py index f99a7eb..be8a4d4 100644 --- a/opennms_api_wrapper/_ipinterfaces_v2.py +++ b/opennms_api_wrapper/_ipinterfaces_v2.py @@ -1,8 +1,10 @@ """IP Interfaces REST API v2 – /api/v2/ipinterfaces (read-only).""" +from ._base import _OpenNMSBase +from typing import Optional -class IpInterfacesV2Mixin: - def get_ip_interfaces(self, fiql: str = None, limit: int = 10, +class IpInterfacesV2Mixin(_OpenNMSBase): + def get_ip_interfaces(self, fiql: Optional[str] = None, limit: int = 10, offset: int = 0): """List IP interfaces using the v2 API with optional FIQL filtering. diff --git a/opennms_api_wrapper/_javamail_config.py b/opennms_api_wrapper/_javamail_config.py index 59d8d5c..aaab024 100644 --- a/opennms_api_wrapper/_javamail_config.py +++ b/opennms_api_wrapper/_javamail_config.py @@ -1,8 +1,9 @@ """Javamail Configuration REST API – /rest/config/javamail.""" +from ._base import _OpenNMSBase from .types import JavamailDefaultConfig, JavamailReadmail, JavamailSendmail, JavamailEnd2End -class JavamailConfigMixin: +class JavamailConfigMixin(_OpenNMSBase): _JM = "config/javamail" # ------------------------------------------------------------------ diff --git a/opennms_api_wrapper/_ksc_reports.py b/opennms_api_wrapper/_ksc_reports.py index 832e967..0e389ff 100644 --- a/opennms_api_wrapper/_ksc_reports.py +++ b/opennms_api_wrapper/_ksc_reports.py @@ -1,8 +1,9 @@ """KSC Reports REST API – /rest/ksc.""" +from ._base import _OpenNMSBase from .types import KscReport -class KscReportsMixin: +class KscReportsMixin(_OpenNMSBase): def get_ksc_reports(self): """List all KSC reports (returns ID and label for each).""" return self._get("ksc") diff --git a/opennms_api_wrapper/_maps.py b/opennms_api_wrapper/_maps.py index 527f641..9ef1eb9 100644 --- a/opennms_api_wrapper/_maps.py +++ b/opennms_api_wrapper/_maps.py @@ -1,8 +1,9 @@ """Maps REST API – /rest/maps.""" +from ._base import _OpenNMSBase from .types import Map -class MapsMixin: +class MapsMixin(_OpenNMSBase): def get_maps(self): """List all maps.""" return self._get("maps") diff --git a/opennms_api_wrapper/_measurements.py b/opennms_api_wrapper/_measurements.py index e23739b..878e064 100644 --- a/opennms_api_wrapper/_measurements.py +++ b/opennms_api_wrapper/_measurements.py @@ -1,8 +1,10 @@ """Measurements REST API – /rest/measurements.""" +from ._base import _OpenNMSBase +from typing import Optional from .types import MeasurementsQuery -class MeasurementsMixin: +class MeasurementsMixin(_OpenNMSBase): def get_measurements( self, resource_id: str, @@ -12,7 +14,7 @@ def get_measurements( step: int = 300000, max_rows: int = 0, aggregation: str = "AVERAGE", - fallback_attribute: str = None, + fallback_attribute: Optional[str] = None, ): """Retrieve time-series values for a single attribute. diff --git a/opennms_api_wrapper/_metadata.py b/opennms_api_wrapper/_metadata.py index 6baf1cd..b27736d 100644 --- a/opennms_api_wrapper/_metadata.py +++ b/opennms_api_wrapper/_metadata.py @@ -1,7 +1,8 @@ """Metadata REST API v2 – /api/v2/nodes/{id}/metadata and sub-resources.""" +from ._base import _OpenNMSBase -class MetadataMixin: +class MetadataMixin(_OpenNMSBase): # ================================================================== # Node metadata # ================================================================== diff --git a/opennms_api_wrapper/_minions.py b/opennms_api_wrapper/_minions.py index 68d51fb..84af3cf 100644 --- a/opennms_api_wrapper/_minions.py +++ b/opennms_api_wrapper/_minions.py @@ -1,7 +1,8 @@ """Minions REST API – /rest/minions.""" +from ._base import _OpenNMSBase -class MinionsMixin: +class MinionsMixin(_OpenNMSBase): def get_minions(self, limit: int = 10, offset: int = 0): """List all minions. diff --git a/opennms_api_wrapper/_monitoring_locations.py b/opennms_api_wrapper/_monitoring_locations.py index 8012fea..4377690 100644 --- a/opennms_api_wrapper/_monitoring_locations.py +++ b/opennms_api_wrapper/_monitoring_locations.py @@ -1,8 +1,9 @@ """Monitoring Locations REST API – /rest/monitoringLocations.""" +from ._base import _OpenNMSBase from .types import MonitoringLocation -class MonitoringLocationsMixin: +class MonitoringLocationsMixin(_OpenNMSBase): def get_monitoring_locations(self, limit: int = 10, offset: int = 0): """List all monitoring locations. diff --git a/opennms_api_wrapper/_nodes.py b/opennms_api_wrapper/_nodes.py index 2851811..c2f4f01 100644 --- a/opennms_api_wrapper/_nodes.py +++ b/opennms_api_wrapper/_nodes.py @@ -1,14 +1,17 @@ """Nodes REST API – /rest/nodes and sub-resources.""" +from __future__ import annotations +from ._base import _OpenNMSBase +from typing import Any, Optional from .types import Node, NodeIpInterface, NodeSnmpInterface, NodeAssetRecord, HardwareEntity, Category -class NodesMixin: +class NodesMixin(_OpenNMSBase): # ================================================================== # Nodes # ================================================================== - def get_nodes(self, limit: int = 10, offset: int = 0, order_by: str = None, - order: str = None, **filters): + def get_nodes(self, limit: int = 10, offset: int = 0, order_by: Optional[str] = None, + order: Optional[str] = None, **filters): """List nodes. Args: @@ -19,7 +22,7 @@ def get_nodes(self, limit: int = 10, offset: int = 0, order_by: str = None, **filters: Additional Hibernate query filters passed directly as query parameters (e.g. ``label="myrouter"``, ``category="Production"``). """ - params = {"limit": limit, "offset": offset} + params: dict[str, Any] = {"limit": limit, "offset": offset} if order_by: params["orderBy"] = order_by if order: @@ -95,7 +98,7 @@ def get_node_ip_interfaces(self, node_id, limit: int = 10, offset: int = 0, 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(f"nodes/{node_id}/ipinterfaces", params=params) @@ -174,7 +177,7 @@ def get_node_snmp_interfaces(self, node_id, limit: int = 10, offset: int = 0, 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(f"nodes/{node_id}/snmpinterfaces", params=params) diff --git a/opennms_api_wrapper/_notifications.py b/opennms_api_wrapper/_notifications.py index e84a40b..2362657 100644 --- a/opennms_api_wrapper/_notifications.py +++ b/opennms_api_wrapper/_notifications.py @@ -1,9 +1,12 @@ """Notifications REST API – /rest/notifications.""" +from __future__ import annotations +from ._base import _OpenNMSBase +from typing import Any, Optional -class NotificationsMixin: +class NotificationsMixin(_OpenNMSBase): def get_notifications(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 notifications. Args: @@ -13,7 +16,7 @@ def get_notifications(self, limit: int = 10, offset: int = 0, order: Sort direction: ``"asc"`` or ``"desc"``. **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} if order_by: params["orderBy"] = order_by if order: diff --git a/opennms_api_wrapper/_outages.py b/opennms_api_wrapper/_outages.py index e0c1ad1..a2bc9c3 100644 --- a/opennms_api_wrapper/_outages.py +++ b/opennms_api_wrapper/_outages.py @@ -1,9 +1,12 @@ """Outages REST API – /rest/outages (read-only).""" +from __future__ import annotations +from ._base import _OpenNMSBase +from typing import Any, Optional -class OutagesMixin: - def get_outages(self, limit: int = 10, offset: int = 0, order_by: str = None, - order: str = None, **filters): +class OutagesMixin(_OpenNMSBase): + def get_outages(self, limit: int = 10, offset: int = 0, order_by: Optional[str] = None, + order: Optional[str] = None, **filters): """List outages (read-only). Args: @@ -14,7 +17,7 @@ def get_outages(self, limit: int = 10, offset: int = 0, order_by: str = None, **filters: Additional Hibernate query filters passed directly as query parameters (e.g. ``node.label="myrouter"``). """ - params = {"limit": limit, "offset": offset} + params: dict[str, Any] = {"limit": limit, "offset": offset} if order_by: params["orderBy"] = order_by if order: diff --git a/opennms_api_wrapper/_perspective_poller.py b/opennms_api_wrapper/_perspective_poller.py index 1fde668..8a72b30 100644 --- a/opennms_api_wrapper/_perspective_poller.py +++ b/opennms_api_wrapper/_perspective_poller.py @@ -1,10 +1,12 @@ """Perspective Poller REST API v2 – /api/v2/perspectivepoller.""" +from ._base import _OpenNMSBase +from typing import Optional -class PerspectivePollerMixin: +class PerspectivePollerMixin(_OpenNMSBase): def get_perspective_poller_status(self, app_id: int, - start: int = None, - end: int = None): + start: Optional[int] = None, + end: Optional[int] = None): """Get perspective poller status for an application. Args: @@ -22,8 +24,8 @@ def get_perspective_poller_status(self, app_id: int, def get_perspective_poller_service_status(self, app_id: int, service_id: int, - start: int = None, - end: int = None): + start: Optional[int] = None, + end: Optional[int] = None): """Get perspective poller status for a specific service. Args: diff --git a/opennms_api_wrapper/_provisiond.py b/opennms_api_wrapper/_provisiond.py index 6d8aea2..e5be118 100644 --- a/opennms_api_wrapper/_provisiond.py +++ b/opennms_api_wrapper/_provisiond.py @@ -1,7 +1,8 @@ """Provisiond REST API v2 – /api/v2/provisiond.""" +from ._base import _OpenNMSBase -class ProvisiondMixin: +class ProvisiondMixin(_OpenNMSBase): def get_provisiond_status(self): """Get the current status of the Provisiond daemon.""" return self._get("provisiond/status", v2=True) diff --git a/opennms_api_wrapper/_requisitions.py b/opennms_api_wrapper/_requisitions.py index 2a528d2..1c07748 100644 --- a/opennms_api_wrapper/_requisitions.py +++ b/opennms_api_wrapper/_requisitions.py @@ -1,8 +1,9 @@ """Requisitions (Provisioning) REST API – /rest/requisitions.""" +from ._base import _OpenNMSBase from .types import RequisitionNode, Requisition -class RequisitionsMixin: +class RequisitionsMixin(_OpenNMSBase): # ================================================================== # Requisitions # ================================================================== diff --git a/opennms_api_wrapper/_resources.py b/opennms_api_wrapper/_resources.py index fda7fca..4b2538f 100644 --- a/opennms_api_wrapper/_resources.py +++ b/opennms_api_wrapper/_resources.py @@ -1,7 +1,9 @@ """Resources REST API – /rest/resources.""" +from ._base import _OpenNMSBase +from typing import Optional -class ResourcesMixin: +class ResourcesMixin(_OpenNMSBase): def get_resources(self, depth: int = 1): """Return the full resource tree (can be expensive on large systems). @@ -33,10 +35,10 @@ def get_resources_for_node(self, node_criteria: str): def get_resources_select( self, - nodes: list = None, - filter_rules: list = None, - node_subresources: list = None, - string_properties: list = None, + nodes: Optional[list] = None, + filter_rules: Optional[list] = None, + node_subresources: Optional[list] = None, + string_properties: Optional[list] = None, ): """Return a partial selection of the resource tree. diff --git a/opennms_api_wrapper/_sched_outages.py b/opennms_api_wrapper/_sched_outages.py index 7b9d257..a531d72 100644 --- a/opennms_api_wrapper/_sched_outages.py +++ b/opennms_api_wrapper/_sched_outages.py @@ -1,8 +1,9 @@ """Scheduled Outages REST API – /rest/sched-outages.""" +from ._base import _OpenNMSBase from .types import SchedOutage -class SchedOutagesMixin: +class SchedOutagesMixin(_OpenNMSBase): # ================================================================== # Scheduled Outages CRUD # ================================================================== diff --git a/opennms_api_wrapper/_scv.py b/opennms_api_wrapper/_scv.py index 1e923f3..81d08f8 100644 --- a/opennms_api_wrapper/_scv.py +++ b/opennms_api_wrapper/_scv.py @@ -1,8 +1,9 @@ """Secure Credentials Vault REST API – /rest/scv.""" +from ._base import _OpenNMSBase from .types import Credential -class ScvMixin: +class ScvMixin(_OpenNMSBase): def get_credentials(self): """List all stored credentials.""" return self._get("scv") diff --git a/opennms_api_wrapper/_situation_feedback.py b/opennms_api_wrapper/_situation_feedback.py index 8137d8b..33e549e 100644 --- a/opennms_api_wrapper/_situation_feedback.py +++ b/opennms_api_wrapper/_situation_feedback.py @@ -1,8 +1,10 @@ """Situation Feedback REST API – /rest/situation-feedback.""" +from ._base import _OpenNMSBase +from typing import Optional -class SituationFeedbackMixin: - def get_situation_feedback_tags(self, prefix: str = None): +class SituationFeedbackMixin(_OpenNMSBase): + def get_situation_feedback_tags(self, prefix: Optional[str] = None): """List situation feedback tags. Args: diff --git a/opennms_api_wrapper/_situations.py b/opennms_api_wrapper/_situations.py index c1403a6..705d3b7 100644 --- a/opennms_api_wrapper/_situations.py +++ b/opennms_api_wrapper/_situations.py @@ -1,7 +1,9 @@ """Situations REST API v2 – /api/v2/situations.""" +from ._base import _OpenNMSBase +from typing import Optional -class SituationsMixin: +class SituationsMixin(_OpenNMSBase): def get_situations(self, limit: int = 10, offset: int = 0): """List situations (v2). @@ -12,8 +14,8 @@ def get_situations(self, limit: int = 10, offset: int = 0): return self._get("situations", params={"limit": limit, "offset": offset}, v2=True) - def create_situation(self, alarm_ids: list, description: str = None, - diagnostic_text: str = None): + def create_situation(self, alarm_ids: list, description: Optional[str] = None, + diagnostic_text: Optional[str] = None): """Create a new situation from a list of alarm IDs. Args: @@ -29,7 +31,7 @@ def create_situation(self, alarm_ids: list, description: str = None, return self._post("situations/create", json_data=body, v2=True) def add_alarms_to_situation(self, situation_id: int, alarm_ids: list, - feedback: str = None): + feedback: Optional[str] = None): """Link additional alarm IDs to an existing situation. Args: diff --git a/opennms_api_wrapper/_snmp_config.py b/opennms_api_wrapper/_snmp_config.py index d994a43..6de9444 100644 --- a/opennms_api_wrapper/_snmp_config.py +++ b/opennms_api_wrapper/_snmp_config.py @@ -1,9 +1,11 @@ """SNMP Configuration REST API – /rest/snmpConfig.""" +from ._base import _OpenNMSBase +from typing import Optional from .types import SnmpConfig -class SnmpConfigMixin: - def get_snmp_config(self, ip_address: str, location: str = None): +class SnmpConfigMixin(_OpenNMSBase): + def get_snmp_config(self, ip_address: str, location: Optional[str] = None): """Get the effective SNMP configuration for *ip_address*. Args: diff --git a/opennms_api_wrapper/_snmp_metadata.py b/opennms_api_wrapper/_snmp_metadata.py index ebec2be..cf9ad7b 100644 --- a/opennms_api_wrapper/_snmp_metadata.py +++ b/opennms_api_wrapper/_snmp_metadata.py @@ -1,7 +1,8 @@ """SNMP Metadata REST API v2 – /api/v2/snmpmetadata.""" +from ._base import _OpenNMSBase -class SnmpMetadataMixin: +class SnmpMetadataMixin(_OpenNMSBase): def get_snmp_metadata(self, node_id: int): """Get SNMP metadata collected for a node. diff --git a/opennms_api_wrapper/_snmpinterfaces_v2.py b/opennms_api_wrapper/_snmpinterfaces_v2.py index 45ccc3b..6ec790c 100644 --- a/opennms_api_wrapper/_snmpinterfaces_v2.py +++ b/opennms_api_wrapper/_snmpinterfaces_v2.py @@ -1,8 +1,10 @@ """SNMP Interfaces REST API v2 – /api/v2/snmpinterfaces (read-only).""" +from ._base import _OpenNMSBase +from typing import Optional -class SnmpInterfacesV2Mixin: - def get_snmp_interfaces(self, fiql: str = None, limit: int = 10, +class SnmpInterfacesV2Mixin(_OpenNMSBase): + def get_snmp_interfaces(self, fiql: Optional[str] = None, limit: int = 10, offset: int = 0): """List SNMP interfaces using the v2 API with optional FIQL filtering. diff --git a/opennms_api_wrapper/_snmptrap_nbi.py b/opennms_api_wrapper/_snmptrap_nbi.py index f6e4ac3..ec932da 100644 --- a/opennms_api_wrapper/_snmptrap_nbi.py +++ b/opennms_api_wrapper/_snmptrap_nbi.py @@ -1,8 +1,9 @@ """SNMP Trap NBI Configuration REST API – /rest/config/snmptrap-nbi.""" +from ._base import _OpenNMSBase from .types import SnmpTrapNbiTrapSink, SnmpTrapNbiConfig -class SnmpTrapNbiMixin: +class SnmpTrapNbiMixin(_OpenNMSBase): _SNMPTRAP_NBI = "config/snmptrap-nbi" def get_snmptrap_nbi_config(self): diff --git a/opennms_api_wrapper/_syslog_nbi.py b/opennms_api_wrapper/_syslog_nbi.py index 24a4b82..edf3882 100644 --- a/opennms_api_wrapper/_syslog_nbi.py +++ b/opennms_api_wrapper/_syslog_nbi.py @@ -1,8 +1,9 @@ """Syslog NBI Configuration REST API – /rest/config/syslog-nbi.""" +from ._base import _OpenNMSBase from .types import SyslogNbiDestination, SyslogNbiConfig -class SyslogNbiMixin: +class SyslogNbiMixin(_OpenNMSBase): _SYSLOG_NBI = "config/syslog-nbi" def get_syslog_nbi_config(self): diff --git a/opennms_api_wrapper/_user_defined_links.py b/opennms_api_wrapper/_user_defined_links.py index 2e50866..e8712a7 100644 --- a/opennms_api_wrapper/_user_defined_links.py +++ b/opennms_api_wrapper/_user_defined_links.py @@ -1,8 +1,9 @@ """User Defined Links REST API v2 – /api/v2/userdefinedlinks.""" +from ._base import _OpenNMSBase from .types import UserDefinedLink -class UserDefinedLinksMixin: +class UserDefinedLinksMixin(_OpenNMSBase): def get_user_defined_links(self): """List all user-defined links.""" return self._get("userdefinedlinks", v2=True) diff --git a/opennms_api_wrapper/_users.py b/opennms_api_wrapper/_users.py index 011e222..c1aceb1 100644 --- a/opennms_api_wrapper/_users.py +++ b/opennms_api_wrapper/_users.py @@ -1,8 +1,9 @@ """Users REST API – /rest/users.""" +from ._base import _OpenNMSBase from .types import User -class UsersMixin: +class UsersMixin(_OpenNMSBase): def get_users(self): """List all users.""" return self._get("users") diff --git a/opennms_api_wrapper/client.py b/opennms_api_wrapper/client.py index 1d64840..2a5b907 100644 --- a/opennms_api_wrapper/client.py +++ b/opennms_api_wrapper/client.py @@ -1,5 +1,4 @@ """Main OpenNMS client class.""" -from ._base import _OpenNMSBase from ._alarms import AlarmsMixin from ._alarm_stats import AlarmStatsMixin from ._alarm_history import AlarmHistoryMixin @@ -59,7 +58,6 @@ class OpenNMS( - _OpenNMSBase, PaginationMixin, InfoMixin, AlarmsMixin, diff --git a/pyproject.toml b/pyproject.toml index 80d8802..6375ed6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -36,7 +36,7 @@ dependencies = [ ] [project.optional-dependencies] -dev = ["pytest", "pytest-cov", "responses", "mkdocs-material", "mkdocstrings[python]", "ruff", "mypy"] +dev = ["pytest", "pytest-cov", "responses", "mkdocs-material", "mkdocstrings[python]", "ruff", "mypy", "types-requests"] [project.urls] Homepage = "https://opennms-api-wrapper.readthedocs.io/en/stable/"