diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..a749376 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,92 @@ +# CLAUDE.md + +This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. + +## Project Overview + +PyCentral is a Python SDK (v2 pre-release) for HPE Aruba Networking Central REST APIs. It supports three platforms: +- **New Central** — latest platform, primary focus of v2 +- **HPE GreenLake Platform (GLP)** — cloud management layer +- **Classic Central** — legacy v1 support in `pycentral/classic/` + +## Common Commands + +### Install for development +```bash +pip install -e . +``` + +### Install documentation dependencies +```bash +pip install -r docs/requirements.txt +``` + +### Build and serve documentation locally +```bash +mkdocs serve +mkdocs build +``` + +### Build distribution packages +```bash +python -m build +``` + +There are no automated tests in this repository. Contributors are expected to test manually before submitting PRs. + +## Architecture + +### Main Entry Point + +`NewCentralBase` (`pycentral/base.py`) is the primary class — it handles connection, OAuth2 token management, and all HTTP requests. It's the only public export from `pycentral/__init__.py`. + +### Module Structure + +- **`pycentral/base.py`** — `NewCentralBase`: HTTP2 client (httpx), OAuth2 flow, automatic token refresh, exponential backoff retry (3 retries, 1–10s), credential loading from YAML/JSON +- **`pycentral/utils/`** — HTTP helpers, auth logic, token management, URL construction; shared across all modules +- **`pycentral/exceptions/`** — Exception hierarchy rooted at `PycentralError`: `LoginError`, `ParameterError`, `ResponseError`, `VerificationError` +- **`pycentral/scopes/`** — Hierarchical infrastructure model: `Scope → SiteCollection → Site → Device/DeviceGroup`; used for bulk operations across device groups +- **`pycentral/profiles/`** — Configuration profile abstraction; supports local (per-scope) and global profiles with a deep-copy modification pattern +- **`pycentral/new_monitoring/`** — Monitoring APIs for devices, APs, clients, gateways, sites on New Central +- **`pycentral/glp/`** — GLP-specific APIs: device management, subscriptions, service manager, user management +- **`pycentral/streaming/`** — WebSocket-based real-time event streaming with Protocol Buffer message definitions (audit, location, geofence, location_analytics) +- **`pycentral/classic/`** — Legacy v1 module (~600KB); all Classic Central API wrappers live here; avoid modifying unless fixing Classic-specific bugs + +### Authentication + +New Central requires `base_url` (or `cluster_name`) + `client_id` + `client_secret`. GLP requires only `client_id` + `client_secret`. Credentials can be passed as a dict or loaded from YAML/JSON files. See `sample_scripts/` for credential file templates. + +## API Documentation + +### Official References +- **New Central Config API:** https://developer.arubanetworks.com/new-central-config/reference +- **New Central API:** https://developer.arubanetworks.com/new-central/reference +- **Getting Started:** https://developer.arubanetworks.com/new-central/docs/getting-started-with-rest-apis + +### Verifying Endpoints + +Append `?json=on` to any reference page URL to retrieve the full OpenAPI 3.1.0 spec for that operation: + +```bash +# Full spec for an operation +curl -s "https://developer.arubanetworks.com/new-central-config/reference/createsite?json=on" + +# Extract request body schema +curl -s "https://developer.arubanetworks.com/new-central-config/reference/createsite?json=on" \ + | jq '.oasDefinition.components.schemas.SitePostRPCInputSchema' + +# Extract endpoint path/method definition +curl -s "https://developer.arubanetworks.com/new-central-config/reference/getsites?json=on" \ + | jq '.oasDefinition.paths."/network-config/v1alpha1/sites".get' +``` + +Always verify exact endpoint paths, required vs. optional query parameters, request body schemas, and response schemas this way before implementing a new API wrapper. + +## Contribution Guidelines + +- All PRs must reference a GitHub issue number and target the **`v2(pre-release)`** branch (not main) +- Commits require DCO sign-off: `git commit -s` +- Code review from at least one maintainer required before merge +- Follow PEP-8; write docstrings in reStructuredText format +- New example scripts go in `sample_scripts/`; library code goes in `pycentral/` +- Test your changes manually before submitting — there is no automated test suite diff --git a/PLAN.md b/PLAN.md new file mode 100644 index 0000000..274deb1 --- /dev/null +++ b/PLAN.md @@ -0,0 +1,506 @@ +# API Coverage Expansion Plan + +This document tracks the plan to implement all unimplemented New Central and New Central Config API +endpoints in pycentral. Each phase is ordered by user impact. + +All PRs implementing items from this plan must: +- Reference a GitHub issue number +- Target `v2(pre-release)` branch +- Include DCO sign-off (`git commit -s`) +- Follow PEP-8 with reStructuredText docstrings + +--- + +## ✅ Prerequisite: `url_utils.py` updates — COMPLETE + +Before or alongside Phase 1, `pycentral/utils/url_utils.py` needs two changes: + +1. **New categories** — add to the `CATEGORIES` dict: + ```python + "notifications": { + "value": "network-notifications", + "type": "monitoring", + "latest": "v1", + }, + "services": { + "value": "network-services", + "type": "monitoring", + "latest": "v1", + }, + "nac": { + "value": "nac-service", + "type": "configuration", + "latest": "v1", + }, + ``` + +2. **Version list** — the `versions` list only contains `["v1alpha1", "v1"]`, but several endpoints + use `v1alpha2` (e.g. gateway monitoring). Add `"v1alpha2"` to the list. + +--- + +## ✅ Phase 1 — High-impact gaps (Monitoring & Services) — COMPLETE + +### 1.1 Switch Monitoring — new file `pycentral/new_monitoring/switches.py` + +New class: `MonitoringSwitches` + +| Method | HTTP | Endpoint | +|---|---|---| +| `get_switches` | GET | `network-monitoring/v1/switches` | +| `get_all_switches` | — | pagination wrapper around `get_switches` | +| `get_switch_details` | GET | `network-monitoring/v1/switches/{serial_number}` | +| `get_switch_ports` | GET | `network-monitoring/v1/switches/{serial_number}/ports` | +| `get_switch_port_details` | GET | `network-monitoring/v1/switches/{serial_number}/ports/{port_id}` | +| `get_switch_port_throughput` | GET | `network-monitoring/v1/switches/{serial_number}/ports/{port_id}/throughput-trends` | +| `get_switch_cpu_utilization` | GET | `network-monitoring/v1/switches/{serial_number}/cpu-utilization-trends` | +| `get_switch_memory_utilization` | GET | `network-monitoring/v1/switches/{serial_number}/memory-utilization-trends` | +| `get_switch_stats` | — | parallel wrapper (cpu + memory) mirroring `MonitoringAPs.get_ap_stats` | +| `get_switch_stacks` | GET | `network-monitoring/v1/switch-stacks` | +| `get_switch_stack_details` | GET | `network-monitoring/v1/switch-stacks/{stack_id}` | +| `get_switch_vlans` | GET | `network-monitoring/v1/switches/{serial_number}/vlans` | + +Verify exact paths and all query parameters from the API reference before implementing: +```bash +curl -s "https://developer.arubanetworks.com/new-central/reference/getswitchesv1?json=on" \ + | jq '.oasDefinition.paths' +``` + +### 1.2 Alerts / Notifications — new file `pycentral/new_monitoring/alerts.py` + +New class: `Alerts` + +Uses `category="notifications"` (new category defined in prerequisite above). + +| Method | HTTP | Endpoint | +|---|---|---| +| `get_alerts` | GET | `network-notifications/v1/alerts` | +| `get_all_alerts` | — | pagination wrapper around `get_alerts` | + +Verify from: +```bash +curl -s "https://developer.arubanetworks.com/new-central/reference/getalertlistv1?json=on" \ + | jq '.oasDefinition.paths' +``` + +### 1.3 Webhooks — new file `pycentral/services/webhooks.py` + +New package `pycentral/services/` with `__init__.py`. New class: `Webhooks` + +Uses `category="services"` (new category defined in prerequisite above). + +| Method | HTTP | Endpoint | +|---|---|---| +| `get_webhooks` | GET | `network-services/v1/webhooks` | +| `create_webhook` | POST | `network-services/v1/webhooks` | +| `get_webhook` | GET | `network-services/v1/webhooks/{webhook_id}` | +| `update_webhook` | PUT | `network-services/v1/webhooks/{webhook_id}` | +| `patch_webhook` | PATCH | `network-services/v1/webhooks/{webhook_id}` | +| `delete_webhook` | DELETE | `network-services/v1/webhooks/{webhook_id}` | +| `rotate_hmac_key` | POST | `network-services/v1/webhooks/{webhook_id}/hmac-key` | + +Verify from: +```bash +curl -s "https://developer.arubanetworks.com/new-central/reference/getwebhooksv1?json=on" \ + | jq '.oasDefinition' +``` + +### 1.4 Device Group Write Operations — extend `pycentral/scopes/` + +Currently `Device_Group` is read-only and `Device_Group.__init__` raises if `from_api=False`. + +Add a standalone `DeviceGroupAPI` class (or functions module) in `pycentral/scopes/device_group_api.py`: + +| Method | HTTP | Endpoint | +|---|---|---| +| `get_device_groups` | GET | `network-config/v1alpha1/device-collections` | +| `create_device_group` | POST | `network-config/v1alpha1/device-collections` | +| `update_device_group` | PUT | `network-config/v1alpha1/device-collections/{scope_id}` | +| `delete_device_group` | DELETE | `network-config/v1alpha1/device-collections/{scope_id}` | +| `delete_device_groups_bulk` | DELETE | `network-config/v1alpha1/device-collections` (bulk) | +| `add_devices` | POST | `network-config/v1alpha1/device-collections/{scope_id}/devices` | +| `remove_devices` | DELETE | `network-config/v1alpha1/device-collections/{scope_id}/devices` | + +Verify from: +```bash +curl -s "https://developer.arubanetworks.com/new-central-config/reference/getdevicegroupsv1?json=on" \ + | jq '.oasDefinition.paths' +``` + +--- + +## ✅ Phase 2 — Services & Supporting Monitoring — COMPLETE + +### 2.1 Firmware Details — `pycentral/services/firmware.py` + +New class: `FirmwareService`. Uses `category="services"`. + +| Method | HTTP | Endpoint | +|---|---|---| +| `get_firmware_details` | GET | `network-services/v1/firmware-details` | + +Verify from: +```bash +curl -s "https://developer.arubanetworks.com/new-central/reference/getfirmwaredetailslistv1?json=on" \ + | jq '.oasDefinition.paths' +``` + +### 2.2 Audit Trail — `pycentral/services/audit.py` + +New class: `AuditTrail`. Uses `category="services"`. + +| Method | HTTP | Endpoint | +|---|---|---| +| `get_audit_events` | GET | `network-services/v1/audit` | +| `get_audit_event_details` | GET | `network-services/v1/audit/{event_id}` | + +Verify from: +```bash +curl -s "https://developer.arubanetworks.com/new-central/reference/get_audit_resp_info?json=on" \ + | jq '.oasDefinition.paths' +``` + +### 2.3 Reporting — `pycentral/new_monitoring/reporting.py` + +New class: `Reporting`. + +| Method | HTTP | Endpoint | +|---|---|---| +| `list_reports` | GET | `network-monitoring/v1alpha1/reports` | +| `create_report` | POST | `network-monitoring/v1alpha1/reports` | +| `get_report` | GET | `network-monitoring/v1alpha1/reports/{report_id}` | +| `update_report` | PUT | `network-monitoring/v1alpha1/reports/{report_id}` | +| `delete_report` | DELETE | `network-monitoring/v1alpha1/reports/{report_id}` | +| `list_report_runs` | GET | `network-monitoring/v1alpha1/reports/{report_id}/runs` | + +Verify from: +```bash +curl -s "https://developer.arubanetworks.com/new-central/reference/listreports?json=on" \ + | jq '.oasDefinition.paths' +``` + +### 2.4 Location Services — `pycentral/new_monitoring/location.py` + +New class: `LocationServices`. + +| Method | HTTP | Endpoint | +|---|---|---| +| `get_device_locations` | GET | `network-monitoring/v1alpha2/device-locations` | +| `get_location_by_id` | GET | `network-monitoring/v1alpha2/locations/{location_id}` | +| `get_device_detailed_location` | GET | `network-monitoring/v1alpha2/devices/{serial_number}/location` | +| `get_ap_ranging_scans` | GET | `network-monitoring/v1alpha2/aps/{serial_number}/ranging-scans` | +| `get_ap_ranging_scan` | GET | `network-monitoring/v1alpha2/aps/{serial_number}/ranging-scans/{scan_id}` | +| `list_asset_tag_data` | GET | `network-monitoring/v1alpha2/asset-tags` | + +Verify all paths and versions from: +```bash +curl -s "https://developer.arubanetworks.com/new-central/reference/getdevicelocationsv1alpha2?json=on" \ + | jq '.oasDefinition.paths' +``` + +### 2.5 Location Analytics — `pycentral/new_monitoring/location_analytics.py` + +New class: `LocationAnalytics`. + +| Method | HTTP | Endpoint | +|---|---|---| +| `get_trends` | GET | `network-monitoring/v1/location-analytics/trends` | +| `get_site_insights` | GET | `network-monitoring/v1/location-analytics/sites/{site_id}/insights` | + +Verify from: +```bash +curl -s "https://developer.arubanetworks.com/new-central/reference/getlatrendsforapiv1?json=on" \ + | jq '.oasDefinition.paths' +``` + +--- + +## ✅ Phase 3 — Troubleshooting Expansion — COMPLETE (pre-existing) + +The existing `pycentral/troubleshooting/troubleshooting.py` only covers the events endpoint. Extend it +(or add sub-modules) with the following. Verify all paths and request schemas before implementing — +the troubleshooting API is the most complex in terms of request bodies. + +### 3.1 AP Troubleshooting + +| Method | HTTP | Endpoint | +|---|---|---| +| `ap_ping` | POST | `network-troubleshooting/v1alpha1/aps/{serial_number}/ping` | +| `ap_traceroute` | POST | `network-troubleshooting/v1alpha1/aps/{serial_number}/traceroute` | +| `ap_speedtest` | POST | `network-troubleshooting/v1alpha1/aps/{serial_number}/speedtest` | +| `ap_show_command` | POST | `network-troubleshooting/v1alpha1/aps/{serial_number}/show-command` | +| `ap_reboot` | POST | `network-troubleshooting/v1alpha1/aps/{serial_number}/reboot` | +| `ap_locate` | POST | `network-troubleshooting/v1alpha1/aps/{serial_number}/locate` | +| `ap_disconnect_users` | POST | `network-troubleshooting/v1alpha1/aps/{serial_number}/disconnect-users` | +| `get_task_status` | GET | `network-troubleshooting/v1alpha1/tasks/{task_id}` | + +### 3.2 Switch Troubleshooting (AOS-CX) + +| Method | HTTP | Endpoint | +|---|---|---| +| `switch_ping` | POST | `network-troubleshooting/v1alpha1/switches/{serial_number}/ping` | +| `switch_traceroute` | POST | `network-troubleshooting/v1alpha1/switches/{serial_number}/traceroute` | +| `switch_poe_bounce` | POST | `network-troubleshooting/v1alpha1/switches/{serial_number}/poe-bounce` | +| `switch_port_bounce` | POST | `network-troubleshooting/v1alpha1/switches/{serial_number}/port-bounce` | +| `switch_cable_test` | POST | `network-troubleshooting/v1alpha1/switches/{serial_number}/cable-test` | +| `switch_show_command` | POST | `network-troubleshooting/v1alpha1/switches/{serial_number}/show-command` | +| `switch_reboot` | POST | `network-troubleshooting/v1alpha1/switches/{serial_number}/reboot` | +| `switch_locate` | POST | `network-troubleshooting/v1alpha1/switches/{serial_number}/locate` | + +### 3.3 Gateway Troubleshooting + +| Method | HTTP | Endpoint | +|---|---|---| +| `gateway_ping` | POST | `network-troubleshooting/v1alpha1/gateways/{serial_number}/ping` | +| `gateway_traceroute` | POST | `network-troubleshooting/v1alpha1/gateways/{serial_number}/traceroute` | +| `gateway_iperf` | POST | `network-troubleshooting/v1alpha1/gateways/{serial_number}/iperf` | +| `gateway_show_command` | POST | `network-troubleshooting/v1alpha1/gateways/{serial_number}/show-command` | +| `gateway_reboot` | POST | `network-troubleshooting/v1alpha1/gateways/{serial_number}/reboot` | +| `gateway_disconnect_clients` | POST | `network-troubleshooting/v1alpha1/gateways/{serial_number}/disconnect-clients` | + +Verify all troubleshooting endpoint paths and body schemas: +```bash +curl -s "https://developer.arubanetworks.com/new-central/reference/appping?json=on" \ + | jq '.oasDefinition' +``` + +--- + +## ✅ Phase 4 — FloorPlan Management — COMPLETE + +New file: `pycentral/new_monitoring/floorplan.py`. New class: `FloorPlan`. +This is a large surface area — implement in sub-groups. + +### 4.1 Floors & Buildings + +| Method | HTTP | Endpoint | +|---|---|---| +| `get_buildings` | GET | `network-monitoring/v1/buildings` | +| `create_floor` | POST | `network-monitoring/v1/floors` | +| `get_floor_summary` | GET | `network-monitoring/v1/floors/{floor_id}` | +| `update_floor_map` | PUT | `network-monitoring/v1/floors/{floor_id}/map` | +| `delete_floor` | DELETE | `network-monitoring/v1/floors/{floor_id}` | +| `get_floor_map_image` | GET | `network-monitoring/v1/floors/{floor_id}/image` | +| `replace_floor_image` | PUT | `network-monitoring/v1/floors/{floor_id}/image` | +| `scale_floor_map` | POST | `network-monitoring/v1/floors/{floor_id}/scale` | +| `import_floors` | POST | `network-monitoring/v1/floors/import` | +| `get_import_status` | GET | `network-monitoring/v1/floors/import/{job_id}` | + +### 4.2 Walls & Zones + +| Method | HTTP | Endpoint | +|---|---|---| +| `get_wall_types` | GET | `network-monitoring/v1/wall-types` | +| `create_wall_types` | POST | `network-monitoring/v1/wall-types` | +| `update_wall_types` | PUT | `network-monitoring/v1/wall-types/{type_id}` | +| `delete_wall_types` | DELETE | `network-monitoring/v1/wall-types/{type_id}` | +| `get_walls` | GET | `network-monitoring/v1/floors/{floor_id}/walls` | +| `create_walls` | POST | `network-monitoring/v1/floors/{floor_id}/walls` | +| `update_walls` | PUT | `network-monitoring/v1/floors/{floor_id}/walls/{wall_id}` | +| `delete_walls` | DELETE | `network-monitoring/v1/floors/{floor_id}/walls/{wall_id}` | +| `get_zones` | GET | `network-monitoring/v1/floors/{floor_id}/zones` | +| `create_zones` | POST | `network-monitoring/v1/floors/{floor_id}/zones` | +| `update_zones` | PUT | `network-monitoring/v1/floors/{floor_id}/zones/{zone_id}` | +| `delete_zones` | DELETE | `network-monitoring/v1/floors/{floor_id}/zones/{zone_id}` | + +### 4.3 Device Placement + +| Method | HTTP | Endpoint | +|---|---|---| +| `get_placed_devices` | GET | `network-monitoring/v1/floors/{floor_id}/devices` | +| `place_devices` | POST | `network-monitoring/v1/floors/{floor_id}/devices` | +| `remove_devices` | DELETE | `network-monitoring/v1/floors/{floor_id}/devices` | +| `change_device_assignment` | PATCH | `network-monitoring/v1/floors/{floor_id}/devices/{device_id}` | +| `get_associated_devices` | GET | `network-monitoring/v1/floors/{floor_id}/associated-devices` | +| `get_heatmap` | GET | `network-monitoring/v1/floors/{floor_id}/heatmap` | +| `get_channel_occupancy_heatmap` | GET | `network-monitoring/v1/floors/{floor_id}/channel-occupancy-heatmap` | + +Verify all FloorPlan paths from: +```bash +curl -s "https://developer.arubanetworks.com/new-central/reference/getsummaryv1?json=on" \ + | jq '.oasDefinition.paths' +``` + +--- + +## Phase 5 — Configuration Profiles (Config API) + +The generic `Profiles` class in `pycentral/profiles/profiles.py` can already call any +`network-config/v1alpha1/` path if you know the endpoint slug. Phase 5 is about adding +**typed, documented wrapper classes** for the major profile categories so callers don't +need to know the raw paths. + +Each wrapper should follow this pattern (see `profiles.py` for the base implementation): +- Accept a `central_conn` and the profile-specific fields +- Delegate CRUD to `Profiles.create_profile` / `get_profile` / `update_profile` / `delete_profile` +- Document all fields and constraints from the OpenAPI spec + +### 5.1 Interface Profiles — new file `pycentral/config/interfaces.py` + +Typed wrappers for: Ethernet Interface, VLAN Interface, Port Channel, LACP, Loopback, +Sub-interfaces, LLDP, CDP, Switch Port Profile, AP Port Profile, sFlow, UFD, Management Interface. + +Verify paths from: +```bash +curl -s "https://developer.arubanetworks.com/new-central-config/reference/readethernetinterfacebyid?json=on" \ + | jq '.oasDefinition.paths' +``` + +### 5.2 Routing & Overlay Profiles — new file `pycentral/config/routing.py` + +Typed wrappers for: Static Route, OSPFv2, OSPFv3, BGP, VRF, Route Map, Prefix List, +AS-Path List, Community List, BFD, PIM, EVPN, Multicast, MSDP, MGMD, Track Object. + +### 5.3 Security Profiles — new file `pycentral/config/security.py` + +Typed wrappers for: AAA Profile, Auth Server, Auth Server Group, AAA Dot1x Auth/Supplicant, +AAA MAC Auth, AAA Captive Portal, Firewall, Certificates, MACsec, MACSec MKA, Port Security, +CoPP, UBT, Auth Survivability. + +### 5.4 Network Service Profiles — new file `pycentral/config/network_services.py` + +Typed wrappers for: DHCP Server, DHCP Pool, DHCP Relay, DHCP Snooping, QoS Global/Queue/Schedule, +Dynamic ARP Inspection, IP Lockdown, ND Snooping, UDP Broadcast Forwarder, NAE Lite, MGMD. + +### 5.5 Roles & Policy Profiles — new file `pycentral/config/policy.py` + +Typed wrappers for: Role, Policy, Object Group, Policy Group, Role ACL, Role GPID, Net Group. + +### 5.6 Named Objects — new file `pycentral/config/named_objects.py` + +Typed wrappers for: Network Service Object, Alias. + +### 5.7 Other Config Profiles + +- Firmware Compliance — `pycentral/config/firmware_policy.py` +- Gateway Clustering / HA — `pycentral/config/gateway_clustering.py` +- Overlay WLAN — `pycentral/config/overlay_wlan.py` +- Application Recognition Control (ARC) — `pycentral/config/arc.py` +- Config Checkpoint — `pycentral/config/checkpoints.py` +- Config Health — add to existing monitoring or new `pycentral/config/health.py` + - `GET network-config/v1alpha1/active-issues` + - `GET network-config/v1alpha1/config-health-devices` + +New `pycentral/config/` package requires a `__init__.py`. + +--- + +## Phase 6 — Central NAC Service + +New package: `pycentral/nac/` with `__init__.py`. New categories needed in `url_utils.py`: +verify the correct base URL prefix from: +```bash +curl -s "https://developer.arubanetworks.com/new-central-config/reference/getmacregistrations?json=on" \ + | jq '.oasDefinition.servers' +``` + +### 6.1 MAC Registration — `pycentral/nac/mac_registration.py` + +| Method | HTTP | Notes | +|---|---|---| +| `get_mac_registrations` | GET | | +| `create_mac_registration` | POST | | +| `update_mac_registration` | PUT | | +| `delete_mac_registration` | DELETE | | +| `export_mac_csv` | GET | streams CSV file | +| `import_mac_csv` | POST | multipart upload | + +### 6.2 Named MPSK — `pycentral/nac/mpsk.py` + +| Method | HTTP | Notes | +|---|---|---| +| `get_named_mpsk` | GET | | +| `create_named_mpsk` | POST | | +| `update_named_mpsk` | PUT | | +| `delete_named_mpsk` | DELETE | | +| `export_mpsk_csv` | GET | streams CSV | +| `import_mpsk_csv` | POST | multipart upload | + +### 6.3 Visitor Management — `pycentral/nac/visitors.py` + +| Method | HTTP | Notes | +|---|---|---| +| `get_visitors` | GET | | +| `create_visitor` | POST | | +| `update_visitor` | PUT | | +| `delete_visitor` | DELETE | | +| `export_visitor_csv` | GET | streams CSV | + +### 6.4 DPP Registration — `pycentral/nac/dpp.py` + +| Method | HTTP | Notes | +|---|---|---| +| `list_dpp_registrations` | GET | | +| `create_dpp_registration` | POST | | +| `update_dpp_registration` | PUT | | +| `get_dpp_registration` | GET | | +| `delete_dpp_registration` | DELETE | | + +### 6.5 Certificates — `pycentral/nac/certificates.py` + +| Method | HTTP | Notes | +|---|---|---| +| `list_certificates` | GET | | +| `revoke_certificates` | POST | | + +### 6.6 NAC Jobs — `pycentral/nac/jobs.py` + +| Method | HTTP | Notes | +|---|---|---| +| `list_jobs` | GET | | +| `get_job` | GET | | +| `delete_job` | DELETE | | +| `download_result_file` | GET | streamed | +| `download_error_file` | GET | streamed | +| `download_input_file` | GET | streamed | + +--- + +## Implementation Notes + +### Adding a new monitoring module + +1. Create `pycentral/new_monitoring/.py` +2. Import `execute_get` from `..utils.monitoring_utils` +3. For write operations, use `central_conn.command("POST"/"PUT"/"DELETE", path, ...)` directly + (see how `pycentral/scopes/site.py` handles create/update/delete) +4. Use `generate_url(endpoint, "monitoring")` for URL construction +5. For `v1alpha2` endpoints, pass `version="v1alpha2"` to `execute_get` or `generate_url` + (requires adding `"v1alpha2"` to `versions` list in `url_utils.py`) + +### Adding a new service module + +1. Create `pycentral/services/.py` +2. Add `"services"` and/or `"notifications"` categories to `url_utils.CATEGORIES` +3. Use `generate_url(endpoint, "services")` for URL construction + +### Adding a new config module + +1. Create `pycentral/config/.py` +2. Import and delegate to `Profiles` from `pycentral.profiles.profiles` +3. Document the profile-specific fields from the OpenAPI spec + +### Docstring format + +All methods must use reStructuredText docstrings. Include the API endpoint in the docstring: + +```python +def get_switches(central_conn, filter_str=None, limit=100, next_page=1): + """ + Retrieve a page of switches. + + This method makes an API call to the following endpoint - ``GET network-monitoring/v1/switches`` + + :param central_conn: Central connection object. + :type central_conn: NewCentralBase + :param filter_str: Optional OData filter expression. + :type filter_str: str, optional + :param limit: Number of entries to return (max 100). + :type limit: int, optional + :param next_page: Pagination cursor (default 1). + :type next_page: int, optional + :return: API response dict with 'items', 'total', 'next'. + :rtype: dict + :raises ParameterError: If limit exceeds 100 or next_page < 1. + """ +``` diff --git a/pycentral/new_monitoring/__init__.py b/pycentral/new_monitoring/__init__.py index a3a46c0..6437019 100644 --- a/pycentral/new_monitoring/__init__.py +++ b/pycentral/new_monitoring/__init__.py @@ -3,3 +3,9 @@ from .gateways import MonitoringGateways from .sites import MonitoringSites from .clients import Clients +from .switches import MonitoringSwitches +from .alerts import Alerts +from .reporting import Reporting +from .location import LocationServices +from .location_analytics import LocationAnalytics +from .floorplan import FloorPlan diff --git a/pycentral/new_monitoring/alerts.py b/pycentral/new_monitoring/alerts.py new file mode 100644 index 0000000..04cfe70 --- /dev/null +++ b/pycentral/new_monitoring/alerts.py @@ -0,0 +1,158 @@ +from ..utils.url_utils import generate_url +from ..utils.monitoring_utils import build_timestamp_filter +from ..exceptions import ParameterError + +ALERT_LIMIT = 100 + + +class Alerts: + @staticmethod + def get_all_alerts( + central_conn, + filter_str=None, + sort=None, + severity=None, + start_time=None, + end_time=None, + duration=None, + ): + """Retrieve all alerts, handling pagination automatically. + + Args: + central_conn (NewCentralBase): Central connection object. + filter_str (str, optional): OData filter expression. + sort (str, optional): Sort expression. + severity (str, optional): Filter by severity (e.g. 'Critical', 'Major', + 'Minor', 'Warning'). + start_time (str, optional): Start time as RFC3339 string or Unix epoch. + end_time (str, optional): End time as RFC3339 string or Unix epoch. + duration (str, optional): Relative duration (e.g. '3h', '1d'). + + Returns: + (list): All alert items across all pages. + """ + alerts = [] + total = None + next_page = 1 + while True: + resp = Alerts.get_alerts( + central_conn, + filter_str=filter_str, + sort=sort, + severity=severity, + limit=ALERT_LIMIT, + next_page=next_page, + start_time=start_time, + end_time=end_time, + duration=duration, + ) + if total is None: + total = resp.get("total", 0) + alerts.extend(resp.get("items", [])) + if len(alerts) >= total: + break + next_val = resp.get("next") + if not next_val: + break + next_page = int(next_val) + return alerts + + @staticmethod + def get_alerts( + central_conn, + filter_str=None, + sort=None, + severity=None, + limit=ALERT_LIMIT, + next_page=1, + start_time=None, + end_time=None, + duration=None, + ): + """Retrieve a page of alerts. + + This method makes an API call to the following endpoint - + ``GET network-notifications/v1/alerts`` + + Args: + central_conn (NewCentralBase): Central connection object. + filter_str (str, optional): OData filter expression. + sort (str, optional): Sort expression. + severity (str, optional): Filter by severity (e.g. 'Critical', 'Major', + 'Minor', 'Warning'). + limit (int, optional): Max results per page (default 100, max 100). + next_page (int, optional): Pagination cursor (default 1). + start_time (str, optional): Start time as RFC3339 string or Unix epoch. + end_time (str, optional): End time as RFC3339 string or Unix epoch. + duration (str, optional): Relative duration (e.g. '3h', '1d'). + + Returns: + (dict): API response containing 'items', 'total', and 'next' keys. + + Raises: + ParameterError: If central_conn is None, limit exceeds ALERT_LIMIT, + or next_page is less than 1. + """ + if not central_conn: + raise ParameterError("central_conn is required") + if limit > ALERT_LIMIT: + raise ParameterError(f"limit cannot exceed {ALERT_LIMIT}") + if next_page < 1: + raise ParameterError("next_page must be 1 or greater") + + params = { + "filter": filter_str, + "sort": sort, + "severity": severity, + "limit": limit, + "next": next_page, + } + + if start_time is not None or end_time is not None or duration is not None: + start_rfc, end_rfc = build_timestamp_filter( + start_time=start_time, + end_time=end_time, + duration=duration, + fmt="rfc3339", + ) + params["start-at"] = start_rfc + params["end-at"] = end_rfc + + path = generate_url("alerts", "notifications", "v1") + resp = central_conn.command("GET", path, api_params=params) + if resp["code"] != 200: + raise Exception( + f"Error retrieving alerts from {path}: {resp['code']} - {resp['msg']}" + ) + return resp["msg"] + + @staticmethod + def get_alert_details(central_conn, alert_id): + """Retrieve details for a specific alert. + + This method makes an API call to the following endpoint - + ``GET network-notifications/v1/alerts/{alert_id}`` + + Args: + central_conn (NewCentralBase): Central connection object. + alert_id (str): Alert identifier. + + Returns: + (dict): Alert details as returned by the API. + + Raises: + ParameterError: If central_conn is None or alert_id is missing or + not a string. + """ + if not central_conn: + raise ParameterError("central_conn is required") + if not alert_id or not isinstance(alert_id, str): + raise ParameterError("alert_id is required and must be a string") + + path = generate_url(f"alerts/{alert_id}", "notifications", "v1") + resp = central_conn.command("GET", path, api_params={}) + if resp["code"] != 200: + raise Exception( + f"Error retrieving alert {alert_id}: {resp['code']} - {resp['msg']}" + ) + return resp["msg"] diff --git a/pycentral/new_monitoring/floorplan.py b/pycentral/new_monitoring/floorplan.py new file mode 100644 index 0000000..37ef500 --- /dev/null +++ b/pycentral/new_monitoring/floorplan.py @@ -0,0 +1,767 @@ +from ..utils.monitoring_utils import execute_get +from ..utils.url_utils import generate_url +from ..exceptions import ParameterError + +FLOOR_LIMIT = 100 + + +class FloorPlan: + """Manage floor plans, buildings, walls, zones, and device placement.""" + + # ------------------------------------------------------------------------- + # Buildings + # ------------------------------------------------------------------------- + + @staticmethod + def get_buildings(central_conn, filter_str=None, sort=None, limit=FLOOR_LIMIT, next_page=1): + """ + Retrieve a list of buildings. + + This method makes an API call to the following endpoint - `GET network-monitoring/v1/buildings` + + Args: + central_conn (NewCentralBase): Central connection object. + filter_str (str, optional): OData filter expression. + sort (str, optional): Sort expression. + limit (int, optional): Max results per page (default 100). + next_page (int, optional): Pagination cursor (default 1). + + Returns: + (dict): API response with building items. + + Raises: + ParameterError: If central_conn is None or limit/next_page invalid. + """ + if not central_conn: + raise ParameterError("central_conn is required") + if limit > FLOOR_LIMIT: + raise ParameterError(f"limit cannot exceed {FLOOR_LIMIT}") + if next_page < 1: + raise ParameterError("next_page must be 1 or greater") + params = {"filter": filter_str, "sort": sort, "limit": limit, "next": next_page} + return execute_get(central_conn, endpoint="buildings", params=params) + + @staticmethod + def update_building(central_conn, building_id, **kwargs): + """ + Update a building. + + This method makes an API call to the following endpoint - `PUT network-monitoring/v1/buildings/{building_id}` + + Args: + central_conn (NewCentralBase): Central connection object. + building_id (str): Building identifier. + **kwargs: Fields to update per API spec (e.g. name, address, campus_id). + + Returns: + (dict): API response. + + Raises: + ParameterError: If building_id is missing. + """ + FloorPlan._validate_str_param(building_id, "building_id") + path = generate_url(f"buildings/{building_id}", "monitoring", "v1") + resp = central_conn.command(api_method="PUT", api_path=path, api_data=kwargs) + if resp["code"] != 200: + raise Exception(f"Error updating building {building_id}: {resp['code']} - {resp['msg']}") + return resp["msg"] + + @staticmethod + def delete_building(central_conn, building_id): + """ + Delete a building. + + This method makes an API call to the following endpoint - `DELETE network-monitoring/v1/buildings/{building_id}` + + Args: + central_conn (NewCentralBase): Central connection object. + building_id (str): Building identifier. + + Returns: + (bool): True if deleted successfully. + + Raises: + ParameterError: If building_id is missing. + """ + FloorPlan._validate_str_param(building_id, "building_id") + path = generate_url(f"buildings/{building_id}", "monitoring", "v1") + resp = central_conn.command(api_method="DELETE", api_path=path) + if resp["code"] != 200: + raise Exception(f"Error deleting building {building_id}: {resp['code']} - {resp['msg']}") + return True + + # ------------------------------------------------------------------------- + # Floors + # ------------------------------------------------------------------------- + + @staticmethod + def create_floor(central_conn, building_id, floor_name, floor_number, **kwargs): + """ + Create a new floor. + + This method makes an API call to the following endpoint - `POST network-monitoring/v1/floors` + + Args: + central_conn (NewCentralBase): Central connection object. + building_id (str): Building identifier to attach the floor to. + floor_name (str): Name of the floor. + floor_number (int): Floor number. + **kwargs: Additional floor fields per API spec (e.g. dimensions, units). + + Returns: + (dict): Created floor response. + + Raises: + ParameterError: If building_id, floor_name, or floor_number is missing. + """ + FloorPlan._validate_str_param(building_id, "building_id") + FloorPlan._validate_str_param(floor_name, "floor_name") + if floor_number is None: + raise ParameterError("floor_number is required") + body = {"building_id": building_id, "floor_name": floor_name, "floor_number": floor_number, **kwargs} + path = generate_url("floors", "monitoring", "v1") + resp = central_conn.command(api_method="POST", api_path=path, api_data=body) + if resp["code"] not in (200, 201): + raise Exception(f"Error creating floor: {resp['code']} - {resp['msg']}") + return resp["msg"] + + @staticmethod + def get_floor(central_conn, floor_id): + """ + Retrieve floor summary. + + This method makes an API call to the following endpoint - `GET network-monitoring/v1/floors/{floor_id}` + + Args: + central_conn (NewCentralBase): Central connection object. + floor_id (str): Floor identifier. + + Returns: + (dict): Floor details. + + Raises: + ParameterError: If floor_id is missing. + """ + FloorPlan._validate_str_param(floor_id, "floor_id") + return execute_get(central_conn, endpoint=f"floors/{floor_id}") + + @staticmethod + def delete_floor(central_conn, floor_id): + """ + Delete a floor. + + This method makes an API call to the following endpoint - `DELETE network-monitoring/v1/floors/{floor_id}` + + Args: + central_conn (NewCentralBase): Central connection object. + floor_id (str): Floor identifier. + + Returns: + (bool): True if deleted successfully. + + Raises: + ParameterError: If floor_id is missing. + """ + FloorPlan._validate_str_param(floor_id, "floor_id") + path = generate_url(f"floors/{floor_id}", "monitoring", "v1") + resp = central_conn.command(api_method="DELETE", api_path=path) + if resp["code"] != 200: + raise Exception(f"Error deleting floor {floor_id}: {resp['code']} - {resp['msg']}") + return True + + @staticmethod + def update_floor_map(central_conn, floor_id, **kwargs): + """ + Update floor map configuration. + + This method makes an API call to the following endpoint - `PUT network-monitoring/v1/floors/{floor_id}/map` + + Args: + central_conn (NewCentralBase): Central connection object. + floor_id (str): Floor identifier. + **kwargs: Map fields per API spec (e.g. width, length, units, rotation). + + Returns: + (dict): API response. + + Raises: + ParameterError: If floor_id is missing. + """ + FloorPlan._validate_str_param(floor_id, "floor_id") + path = generate_url(f"floors/{floor_id}/map", "monitoring", "v1") + resp = central_conn.command(api_method="PUT", api_path=path, api_data=kwargs) + if resp["code"] != 200: + raise Exception(f"Error updating floor map {floor_id}: {resp['code']} - {resp['msg']}") + return resp["msg"] + + @staticmethod + def scale_floor_map(central_conn, floor_id, **kwargs): + """ + Scale the floor map. + + This method makes an API call to the following endpoint - `POST network-monitoring/v1/floors/{floor_id}/scale` + + Args: + central_conn (NewCentralBase): Central connection object. + floor_id (str): Floor identifier. + **kwargs: Scale parameters per API spec (e.g. pixels_per_unit, scale_type). + + Returns: + (dict): API response. + + Raises: + ParameterError: If floor_id is missing. + """ + FloorPlan._validate_str_param(floor_id, "floor_id") + path = generate_url(f"floors/{floor_id}/scale", "monitoring", "v1") + resp = central_conn.command(api_method="POST", api_path=path, api_data=kwargs) + if resp["code"] not in (200, 201): + raise Exception(f"Error scaling floor map {floor_id}: {resp['code']} - {resp['msg']}") + return resp["msg"] + + @staticmethod + def import_floors(central_conn, floor_data): + """ + Import floors in bulk. + + This method makes an API call to the following endpoint - `POST network-monitoring/v1/floors/import` + + Args: + central_conn (NewCentralBase): Central connection object. + floor_data (dict): Import payload per API spec. + + Returns: + (dict): API response including a job_id for status polling. + + Raises: + ParameterError: If floor_data is missing. + """ + if not floor_data or not isinstance(floor_data, dict): + raise ParameterError("floor_data is required and must be a dict") + path = generate_url("floors/import", "monitoring", "v1") + resp = central_conn.command(api_method="POST", api_path=path, api_data=floor_data) + if resp["code"] not in (200, 201, 202): + raise Exception(f"Error importing floors: {resp['code']} - {resp['msg']}") + return resp["msg"] + + @staticmethod + def get_import_status(central_conn, job_id): + """ + Get the status of a floor import job. + + This method makes an API call to the following endpoint - `GET network-monitoring/v1/floors/import/{job_id}` + + Args: + central_conn (NewCentralBase): Central connection object. + job_id (str): Import job identifier. + + Returns: + (dict): Import job status. + + Raises: + ParameterError: If job_id is missing. + """ + FloorPlan._validate_str_param(job_id, "job_id") + return execute_get(central_conn, endpoint=f"floors/import/{job_id}") + + # ------------------------------------------------------------------------- + # Wall Types + # ------------------------------------------------------------------------- + + @staticmethod + def get_wall_types(central_conn): + """ + Retrieve all wall types. + + This method makes an API call to the following endpoint - `GET network-monitoring/v1/wall-types` + + Args: + central_conn (NewCentralBase): Central connection object. + + Returns: + (dict): Wall types response. + + Raises: + ParameterError: If central_conn is None. + """ + if not central_conn: + raise ParameterError("central_conn is required") + return execute_get(central_conn, endpoint="wall-types") + + @staticmethod + def create_wall_types(central_conn, wall_types): + """ + Create one or more wall types. + + This method makes an API call to the following endpoint - `POST network-monitoring/v1/wall-types` + + Args: + central_conn (NewCentralBase): Central connection object. + wall_types (list[dict]): List of wall type objects per API spec. + + Returns: + (dict): API response. + + Raises: + ParameterError: If wall_types is not a non-empty list. + """ + if not wall_types or not isinstance(wall_types, list): + raise ParameterError("wall_types must be a non-empty list") + path = generate_url("wall-types", "monitoring", "v1") + resp = central_conn.command(api_method="POST", api_path=path, api_data={"wall_types": wall_types}) + if resp["code"] not in (200, 201): + raise Exception(f"Error creating wall types: {resp['code']} - {resp['msg']}") + return resp["msg"] + + @staticmethod + def update_wall_type(central_conn, type_id, **kwargs): + """ + Update a wall type. + + This method makes an API call to the following endpoint - `PUT network-monitoring/v1/wall-types/{type_id}` + + Args: + central_conn (NewCentralBase): Central connection object. + type_id (str): Wall type identifier. + **kwargs: Fields to update per API spec. + + Returns: + (dict): API response. + + Raises: + ParameterError: If type_id is missing. + """ + FloorPlan._validate_str_param(type_id, "type_id") + path = generate_url(f"wall-types/{type_id}", "monitoring", "v1") + resp = central_conn.command(api_method="PUT", api_path=path, api_data=kwargs) + if resp["code"] != 200: + raise Exception(f"Error updating wall type {type_id}: {resp['code']} - {resp['msg']}") + return resp["msg"] + + @staticmethod + def delete_wall_type(central_conn, type_id): + """ + Delete a wall type. + + This method makes an API call to the following endpoint - `DELETE network-monitoring/v1/wall-types/{type_id}` + + Args: + central_conn (NewCentralBase): Central connection object. + type_id (str): Wall type identifier. + + Returns: + (bool): True if deleted successfully. + + Raises: + ParameterError: If type_id is missing. + """ + FloorPlan._validate_str_param(type_id, "type_id") + path = generate_url(f"wall-types/{type_id}", "monitoring", "v1") + resp = central_conn.command(api_method="DELETE", api_path=path) + if resp["code"] != 200: + raise Exception(f"Error deleting wall type {type_id}: {resp['code']} - {resp['msg']}") + return True + + # ------------------------------------------------------------------------- + # Walls + # ------------------------------------------------------------------------- + + @staticmethod + def get_walls(central_conn, floor_id): + """ + Retrieve walls for a floor. + + This method makes an API call to the following endpoint - `GET network-monitoring/v1/floors/{floor_id}/walls` + + Args: + central_conn (NewCentralBase): Central connection object. + floor_id (str): Floor identifier. + + Returns: + (dict): Walls response. + + Raises: + ParameterError: If floor_id is missing. + """ + FloorPlan._validate_str_param(floor_id, "floor_id") + return execute_get(central_conn, endpoint=f"floors/{floor_id}/walls") + + @staticmethod + def create_walls(central_conn, floor_id, walls): + """ + Create walls on a floor. + + This method makes an API call to the following endpoint - `POST network-monitoring/v1/floors/{floor_id}/walls` + + Args: + central_conn (NewCentralBase): Central connection object. + floor_id (str): Floor identifier. + walls (list[dict]): List of wall objects per API spec. + + Returns: + (dict): API response. + + Raises: + ParameterError: If floor_id missing or walls is not a non-empty list. + """ + FloorPlan._validate_str_param(floor_id, "floor_id") + if not walls or not isinstance(walls, list): + raise ParameterError("walls must be a non-empty list") + path = generate_url(f"floors/{floor_id}/walls", "monitoring", "v1") + resp = central_conn.command(api_method="POST", api_path=path, api_data={"walls": walls}) + if resp["code"] not in (200, 201): + raise Exception(f"Error creating walls on floor {floor_id}: {resp['code']} - {resp['msg']}") + return resp["msg"] + + @staticmethod + def update_wall(central_conn, floor_id, wall_id, **kwargs): + """ + Update a wall on a floor. + + This method makes an API call to the following endpoint - `PUT network-monitoring/v1/floors/{floor_id}/walls/{wall_id}` + + Args: + central_conn (NewCentralBase): Central connection object. + floor_id (str): Floor identifier. + wall_id (str): Wall identifier. + **kwargs: Fields to update per API spec. + + Returns: + (dict): API response. + + Raises: + ParameterError: If floor_id or wall_id is missing. + """ + FloorPlan._validate_str_param(floor_id, "floor_id") + FloorPlan._validate_str_param(wall_id, "wall_id") + path = generate_url(f"floors/{floor_id}/walls/{wall_id}", "monitoring", "v1") + resp = central_conn.command(api_method="PUT", api_path=path, api_data=kwargs) + if resp["code"] != 200: + raise Exception(f"Error updating wall {wall_id}: {resp['code']} - {resp['msg']}") + return resp["msg"] + + @staticmethod + def delete_wall(central_conn, floor_id, wall_id): + """ + Delete a wall from a floor. + + This method makes an API call to the following endpoint - `DELETE network-monitoring/v1/floors/{floor_id}/walls/{wall_id}` + + Args: + central_conn (NewCentralBase): Central connection object. + floor_id (str): Floor identifier. + wall_id (str): Wall identifier. + + Returns: + (bool): True if deleted successfully. + + Raises: + ParameterError: If floor_id or wall_id is missing. + """ + FloorPlan._validate_str_param(floor_id, "floor_id") + FloorPlan._validate_str_param(wall_id, "wall_id") + path = generate_url(f"floors/{floor_id}/walls/{wall_id}", "monitoring", "v1") + resp = central_conn.command(api_method="DELETE", api_path=path) + if resp["code"] != 200: + raise Exception(f"Error deleting wall {wall_id}: {resp['code']} - {resp['msg']}") + return True + + # ------------------------------------------------------------------------- + # Zones + # ------------------------------------------------------------------------- + + @staticmethod + def get_zones(central_conn, floor_id): + """ + Retrieve zones for a floor. + + This method makes an API call to the following endpoint - `GET network-monitoring/v1/floors/{floor_id}/zones` + + Args: + central_conn (NewCentralBase): Central connection object. + floor_id (str): Floor identifier. + + Returns: + (dict): Zones response. + + Raises: + ParameterError: If floor_id is missing. + """ + FloorPlan._validate_str_param(floor_id, "floor_id") + return execute_get(central_conn, endpoint=f"floors/{floor_id}/zones") + + @staticmethod + def create_zones(central_conn, floor_id, zones): + """ + Create zones on a floor. + + This method makes an API call to the following endpoint - `POST network-monitoring/v1/floors/{floor_id}/zones` + + Args: + central_conn (NewCentralBase): Central connection object. + floor_id (str): Floor identifier. + zones (list[dict]): List of zone objects per API spec. + + Returns: + (dict): API response. + + Raises: + ParameterError: If floor_id missing or zones is not a non-empty list. + """ + FloorPlan._validate_str_param(floor_id, "floor_id") + if not zones or not isinstance(zones, list): + raise ParameterError("zones must be a non-empty list") + path = generate_url(f"floors/{floor_id}/zones", "monitoring", "v1") + resp = central_conn.command(api_method="POST", api_path=path, api_data={"zones": zones}) + if resp["code"] not in (200, 201): + raise Exception(f"Error creating zones on floor {floor_id}: {resp['code']} - {resp['msg']}") + return resp["msg"] + + @staticmethod + def update_zone(central_conn, floor_id, zone_id, **kwargs): + """ + Update a zone on a floor. + + This method makes an API call to the following endpoint - `PUT network-monitoring/v1/floors/{floor_id}/zones/{zone_id}` + + Args: + central_conn (NewCentralBase): Central connection object. + floor_id (str): Floor identifier. + zone_id (str): Zone identifier. + **kwargs: Fields to update per API spec. + + Returns: + (dict): API response. + + Raises: + ParameterError: If floor_id or zone_id is missing. + """ + FloorPlan._validate_str_param(floor_id, "floor_id") + FloorPlan._validate_str_param(zone_id, "zone_id") + path = generate_url(f"floors/{floor_id}/zones/{zone_id}", "monitoring", "v1") + resp = central_conn.command(api_method="PUT", api_path=path, api_data=kwargs) + if resp["code"] != 200: + raise Exception(f"Error updating zone {zone_id}: {resp['code']} - {resp['msg']}") + return resp["msg"] + + @staticmethod + def delete_zone(central_conn, floor_id, zone_id): + """ + Delete a zone from a floor. + + This method makes an API call to the following endpoint - `DELETE network-monitoring/v1/floors/{floor_id}/zones/{zone_id}` + + Args: + central_conn (NewCentralBase): Central connection object. + floor_id (str): Floor identifier. + zone_id (str): Zone identifier. + + Returns: + (bool): True if deleted successfully. + + Raises: + ParameterError: If floor_id or zone_id is missing. + """ + FloorPlan._validate_str_param(floor_id, "floor_id") + FloorPlan._validate_str_param(zone_id, "zone_id") + path = generate_url(f"floors/{floor_id}/zones/{zone_id}", "monitoring", "v1") + resp = central_conn.command(api_method="DELETE", api_path=path) + if resp["code"] != 200: + raise Exception(f"Error deleting zone {zone_id}: {resp['code']} - {resp['msg']}") + return True + + # ------------------------------------------------------------------------- + # Device Placement + # ------------------------------------------------------------------------- + + @staticmethod + def get_placed_devices(central_conn, floor_id, filter_str=None): + """ + Retrieve devices placed on a floor. + + This method makes an API call to the following endpoint - `GET network-monitoring/v1/floors/{floor_id}/devices` + + Args: + central_conn (NewCentralBase): Central connection object. + floor_id (str): Floor identifier. + filter_str (str, optional): OData filter expression. + + Returns: + (dict): Placed devices response. + + Raises: + ParameterError: If floor_id is missing. + """ + FloorPlan._validate_str_param(floor_id, "floor_id") + params = {"filter": filter_str} + return execute_get(central_conn, endpoint=f"floors/{floor_id}/devices", params=params) + + @staticmethod + def place_devices(central_conn, floor_id, devices): + """ + Place devices on a floor. + + This method makes an API call to the following endpoint - `POST network-monitoring/v1/floors/{floor_id}/devices` + + Args: + central_conn (NewCentralBase): Central connection object. + floor_id (str): Floor identifier. + devices (list[dict]): List of device placement objects per API spec. + + Returns: + (dict): API response. + + Raises: + ParameterError: If floor_id missing or devices is not a non-empty list. + """ + FloorPlan._validate_str_param(floor_id, "floor_id") + if not devices or not isinstance(devices, list): + raise ParameterError("devices must be a non-empty list") + path = generate_url(f"floors/{floor_id}/devices", "monitoring", "v1") + resp = central_conn.command(api_method="POST", api_path=path, api_data={"devices": devices}) + if resp["code"] not in (200, 201): + raise Exception(f"Error placing devices on floor {floor_id}: {resp['code']} - {resp['msg']}") + return resp["msg"] + + @staticmethod + def remove_devices(central_conn, floor_id, device_ids): + """ + Remove devices from a floor. + + This method makes an API call to the following endpoint - `DELETE network-monitoring/v1/floors/{floor_id}/devices` + + Args: + central_conn (NewCentralBase): Central connection object. + floor_id (str): Floor identifier. + device_ids (list[str]): List of device IDs to remove. + + Returns: + (bool): True if removed successfully. + + Raises: + ParameterError: If floor_id missing or device_ids not a non-empty list. + """ + FloorPlan._validate_str_param(floor_id, "floor_id") + if not device_ids or not isinstance(device_ids, list): + raise ParameterError("device_ids must be a non-empty list") + path = generate_url(f"floors/{floor_id}/devices", "monitoring", "v1") + resp = central_conn.command(api_method="DELETE", api_path=path, + api_data={"device_ids": device_ids}) + if resp["code"] != 200: + raise Exception(f"Error removing devices from floor {floor_id}: {resp['code']} - {resp['msg']}") + return True + + @staticmethod + def change_device_assignment(central_conn, floor_id, device_id, **kwargs): + """ + Change a device's placement or assignment on a floor. + + This method makes an API call to the following endpoint - `PATCH network-monitoring/v1/floors/{floor_id}/devices/{device_id}` + + Args: + central_conn (NewCentralBase): Central connection object. + floor_id (str): Floor identifier. + device_id (str): Device identifier. + **kwargs: Placement fields per API spec (e.g. x, y, orientation). + + Returns: + (dict): API response. + + Raises: + ParameterError: If floor_id or device_id is missing. + """ + FloorPlan._validate_str_param(floor_id, "floor_id") + FloorPlan._validate_str_param(device_id, "device_id") + path = generate_url(f"floors/{floor_id}/devices/{device_id}", "monitoring", "v1") + resp = central_conn.command(api_method="PATCH", api_path=path, api_data=kwargs) + if resp["code"] != 200: + raise Exception(f"Error changing device {device_id} assignment: {resp['code']} - {resp['msg']}") + return resp["msg"] + + @staticmethod + def get_associated_devices(central_conn, floor_id): + """ + Retrieve devices associated with a floor (not necessarily placed). + + This method makes an API call to the following endpoint - `GET network-monitoring/v1/floors/{floor_id}/associated-devices` + + Args: + central_conn (NewCentralBase): Central connection object. + floor_id (str): Floor identifier. + + Returns: + (dict): Associated devices response. + + Raises: + ParameterError: If floor_id is missing. + """ + FloorPlan._validate_str_param(floor_id, "floor_id") + return execute_get(central_conn, endpoint=f"floors/{floor_id}/associated-devices") + + # ------------------------------------------------------------------------- + # Heatmaps + # ------------------------------------------------------------------------- + + @staticmethod + def get_heatmap(central_conn, floor_id, heatmap_type=None): + """ + Retrieve heatmap data for a floor. + + This method makes an API call to the following endpoint - `GET network-monitoring/v1/floors/{floor_id}/heatmap` + + Args: + central_conn (NewCentralBase): Central connection object. + floor_id (str): Floor identifier. + heatmap_type (str, optional): Type of heatmap (e.g. 'rssi', 'snr'). + + Returns: + (dict): Heatmap data. + + Raises: + ParameterError: If floor_id is missing. + """ + FloorPlan._validate_str_param(floor_id, "floor_id") + params = {"type": heatmap_type} if heatmap_type else {} + return execute_get(central_conn, endpoint=f"floors/{floor_id}/heatmap", params=params) + + @staticmethod + def get_channel_occupancy_heatmap(central_conn, floor_id): + """ + Retrieve channel occupancy heatmap for a floor. + + This method makes an API call to the following endpoint - `GET network-monitoring/v1/floors/{floor_id}/channel-occupancy-heatmap` + + Args: + central_conn (NewCentralBase): Central connection object. + floor_id (str): Floor identifier. + + Returns: + (dict): Channel occupancy heatmap data. + + Raises: + ParameterError: If floor_id is missing. + """ + FloorPlan._validate_str_param(floor_id, "floor_id") + return execute_get(central_conn, endpoint=f"floors/{floor_id}/channel-occupancy-heatmap") + + # ------------------------------------------------------------------------- + # Internal helpers + # ------------------------------------------------------------------------- + + @staticmethod + def _validate_str_param(value, name): + """ + Validate that a parameter is a non-empty string. + + Args: + value: Value to validate. + name (str): Parameter name for the error message. + + Raises: + ParameterError: If value is not a non-empty string. + + Note: + Internal SDK function + """ + if not value or not isinstance(value, str): + raise ParameterError(f"{name} is required and must be a non-empty string") diff --git a/pycentral/new_monitoring/location.py b/pycentral/new_monitoring/location.py new file mode 100644 index 0000000..57d7360 --- /dev/null +++ b/pycentral/new_monitoring/location.py @@ -0,0 +1,194 @@ +from ..utils.monitoring_utils import execute_get +from ..exceptions import ParameterError + +LOCATION_LIMIT = 100 + + +class LocationServices: + + @staticmethod + def get_device_locations(central_conn, filter_str=None, sort=None, limit=LOCATION_LIMIT, next_page=1): + """ + Retrieve location data for all devices. + + This method makes an API call to the following endpoint - `GET network-monitoring/v1alpha2/device-locations` + + Args: + central_conn (NewCentralBase): Central connection object. + filter_str (str, optional): OData filter expression. + sort (str, optional): Sort expression. + limit (int, optional): Max results (default 100). + next_page (int, optional): Pagination cursor (default 1). + + Returns: + (dict): API response with device location items. + + Raises: + ParameterError: If limit or next_page invalid. + """ + if not central_conn: + raise ParameterError("central_conn is required") + if limit > LOCATION_LIMIT: + raise ParameterError(f"limit cannot exceed {LOCATION_LIMIT}") + if next_page < 1: + raise ParameterError("next_page must be 1 or greater") + params = {"filter": filter_str, "sort": sort, "limit": limit, "next": next_page} + return execute_get(central_conn, endpoint="device-locations", params=params, version="v1alpha2") + + @staticmethod + def get_all_device_locations(central_conn, filter_str=None, sort=None): + """Retrieve all device locations, handling pagination. + + Args: + central_conn (NewCentralBase): Central connection object. + filter_str (str, optional): OData filter expression. + sort (str, optional): Sort expression. + + Returns: + (list[dict]): List of all device location items. + """ + locations = [] + total = None + next_page = 1 + while True: + resp = LocationServices.get_device_locations( + central_conn, filter_str=filter_str, sort=sort, + limit=LOCATION_LIMIT, next_page=next_page + ) + if total is None: + total = resp.get("total", 0) + locations.extend(resp.get("items", [])) + if len(locations) >= total: + break + next_val = resp.get("next") + if not next_val: + break + next_page = int(next_val) + return locations + + @staticmethod + def get_location_by_id(central_conn, location_id): + """ + Retrieve a location by its ID. + + This method makes an API call to the following endpoint - `GET network-monitoring/v1alpha2/locations/{location_id}` + + Args: + central_conn (NewCentralBase): Central connection object. + location_id (str): Location identifier. + + Returns: + (dict): Location details. + + Raises: + ParameterError: If location_id missing. + """ + if not location_id or not isinstance(location_id, str): + raise ParameterError("location_id is required and must be a string") + return execute_get(central_conn, endpoint=f"locations/{location_id}", version="v1alpha2") + + @staticmethod + def get_device_detailed_location(central_conn, serial_number): + """ + Retrieve detailed location information for a specific device. + + This method makes an API call to the following endpoint - `GET network-monitoring/v1alpha2/devices/{serial_number}/location` + + Args: + central_conn (NewCentralBase): Central connection object. + serial_number (str): Device serial number. + + Returns: + (dict): Detailed location information. + + Raises: + ParameterError: If serial_number missing. + """ + if not serial_number or not isinstance(serial_number, str): + raise ParameterError("serial_number is required and must be a string") + return execute_get(central_conn, endpoint=f"devices/{serial_number}/location", version="v1alpha2") + + @staticmethod + def get_ap_ranging_scans(central_conn, serial_number, filter_str=None, sort=None, limit=LOCATION_LIMIT, next_page=1): + """ + Retrieve ranging scan list for an AP. + + This method makes an API call to the following endpoint - `GET network-monitoring/v1alpha2/aps/{serial_number}/ranging-scans` + + Args: + central_conn (NewCentralBase): Central connection object. + serial_number (str): AP serial number. + filter_str (str, optional): OData filter expression. + sort (str, optional): Sort expression. + limit (int, optional): Max results (default 100). + next_page (int, optional): Pagination cursor (default 1). + + Returns: + (dict): API response with ranging scan items. + + Raises: + ParameterError: If serial_number missing or limit/next_page invalid. + """ + if not serial_number or not isinstance(serial_number, str): + raise ParameterError("serial_number is required and must be a string") + if limit > LOCATION_LIMIT: + raise ParameterError(f"limit cannot exceed {LOCATION_LIMIT}") + if next_page < 1: + raise ParameterError("next_page must be 1 or greater") + params = {"filter": filter_str, "sort": sort, "limit": limit, "next": next_page} + return execute_get(central_conn, endpoint=f"aps/{serial_number}/ranging-scans", + params=params, version="v1alpha2") + + @staticmethod + def get_ap_ranging_scan(central_conn, serial_number, scan_id): + """ + Retrieve a specific ranging scan for an AP. + + This method makes an API call to the following endpoint - `GET network-monitoring/v1alpha2/aps/{serial_number}/ranging-scans/{scan_id}` + + Args: + central_conn (NewCentralBase): Central connection object. + serial_number (str): AP serial number. + scan_id (str): Ranging scan identifier. + + Returns: + (dict): Ranging scan details. + + Raises: + ParameterError: If serial_number or scan_id missing. + """ + if not serial_number or not isinstance(serial_number, str): + raise ParameterError("serial_number is required and must be a string") + if not scan_id or not isinstance(scan_id, str): + raise ParameterError("scan_id is required and must be a string") + return execute_get(central_conn, endpoint=f"aps/{serial_number}/ranging-scans/{scan_id}", + version="v1alpha2") + + @staticmethod + def list_asset_tag_data(central_conn, filter_str=None, sort=None, limit=LOCATION_LIMIT, next_page=1): + """ + Retrieve asset tag location data. + + This method makes an API call to the following endpoint - `GET network-monitoring/v1alpha2/asset-tags` + + Args: + central_conn (NewCentralBase): Central connection object. + filter_str (str, optional): OData filter expression. + sort (str, optional): Sort expression. + limit (int, optional): Max results (default 100). + next_page (int, optional): Pagination cursor (default 1). + + Returns: + (dict): API response with asset tag items. + + Raises: + ParameterError: If limit or next_page invalid. + """ + if not central_conn: + raise ParameterError("central_conn is required") + if limit > LOCATION_LIMIT: + raise ParameterError(f"limit cannot exceed {LOCATION_LIMIT}") + if next_page < 1: + raise ParameterError("next_page must be 1 or greater") + params = {"filter": filter_str, "sort": sort, "limit": limit, "next": next_page} + return execute_get(central_conn, endpoint="asset-tags", params=params, version="v1alpha2") diff --git a/pycentral/new_monitoring/location_analytics.py b/pycentral/new_monitoring/location_analytics.py new file mode 100644 index 0000000..d76c3a9 --- /dev/null +++ b/pycentral/new_monitoring/location_analytics.py @@ -0,0 +1,70 @@ +from ..utils.monitoring_utils import execute_get, build_timestamp_filter +from ..exceptions import ParameterError + + +class LocationAnalytics: + + @staticmethod + def get_trends(central_conn, site_id=None, start_time=None, end_time=None, duration=None): + """ + Retrieve location analytics trends. + + This method makes an API call to the following endpoint - `GET network-monitoring/v1/location-analytics/trends` + + Args: + central_conn (NewCentralBase): Central connection object. + site_id (str, optional): Site identifier to filter by. + start_time (str, optional): Start time (RFC3339 or epoch). + end_time (str, optional): End time (RFC3339 or epoch). + duration (str, optional): Relative duration (e.g. '3h', '1d'). + + Returns: + (dict): Location analytics trend data. + + Raises: + ParameterError: If central_conn is None. + """ + if not central_conn: + raise ParameterError("central_conn is required") + params = {} + if site_id: + params["site-id"] = site_id + if start_time is not None or end_time is not None or duration is not None: + start_rfc, end_rfc = build_timestamp_filter( + start_time=start_time, end_time=end_time, duration=duration, fmt="rfc3339" + ) + params["start-at"] = start_rfc + params["end-at"] = end_rfc + return execute_get(central_conn, endpoint="location-analytics/trends", params=params) + + @staticmethod + def get_site_insights(central_conn, site_id, start_time=None, end_time=None, duration=None): + """ + Retrieve location analytics insights for a site. + + This method makes an API call to the following endpoint - `GET network-monitoring/v1/location-analytics/sites/{site_id}/insights` + + Args: + central_conn (NewCentralBase): Central connection object. + site_id (str): Site identifier. + start_time (str, optional): Start time (RFC3339 or epoch). + end_time (str, optional): End time (RFC3339 or epoch). + duration (str, optional): Relative duration (e.g. '3h', '1d'). + + Returns: + (dict): Site-level location analytics insights. + + Raises: + ParameterError: If site_id missing. + """ + if not site_id or not isinstance(site_id, str): + raise ParameterError("site_id is required and must be a string") + params = {} + if start_time is not None or end_time is not None or duration is not None: + start_rfc, end_rfc = build_timestamp_filter( + start_time=start_time, end_time=end_time, duration=duration, fmt="rfc3339" + ) + params["start-at"] = start_rfc + params["end-at"] = end_rfc + return execute_get(central_conn, endpoint=f"location-analytics/sites/{site_id}/insights", + params=params) diff --git a/pycentral/new_monitoring/reporting.py b/pycentral/new_monitoring/reporting.py new file mode 100644 index 0000000..7fa796b --- /dev/null +++ b/pycentral/new_monitoring/reporting.py @@ -0,0 +1,203 @@ +# (C) Copyright 2025 Hewlett Packard Enterprise Development LP. +# MIT License + +from ..utils.monitoring_utils import execute_get +from ..utils.url_utils import generate_url +from ..exceptions import ParameterError + +REPORT_LIMIT = 100 +REPORTS_ENDPOINT = "reports" + + +class Reporting: + + @staticmethod + def list_reports(central_conn, filter_str=None, sort=None, limit=REPORT_LIMIT, next_page=1): + """Retrieve a page of saved reports. + + This method makes an API call to the following endpoint - + ``GET network-monitoring/v1alpha1/reports`` + + Args: + central_conn (NewCentralBase): Central connection object. + filter_str (str, optional): OData filter expression. + sort (str, optional): Sort expression. + limit (int, optional): Max results per page (default 100). + next_page (int, optional): Pagination cursor (default 1). + + Returns: + (dict): API response with 'items', 'total', 'next'. + + Raises: + ParameterError: If limit or next_page invalid. + """ + if not central_conn: + raise ParameterError("central_conn is required") + if limit > REPORT_LIMIT: + raise ParameterError(f"limit cannot exceed {REPORT_LIMIT}") + if next_page < 1: + raise ParameterError("next_page must be 1 or greater") + params = {"filter": filter_str, "sort": sort, "limit": limit, "next": next_page} + return execute_get(central_conn, endpoint=REPORTS_ENDPOINT, params=params, version="v1alpha1") + + @staticmethod + def get_all_reports(central_conn, filter_str=None, sort=None): + """Retrieve all saved reports, handling pagination automatically. + + Args: + central_conn (NewCentralBase): Central connection object. + filter_str (str, optional): OData filter expression. + sort (str, optional): Sort expression. + + Returns: + (list): All report items across all pages. + """ + reports = [] + total = None + next_page = 1 + while True: + resp = Reporting.list_reports(central_conn, filter_str=filter_str, sort=sort, + limit=REPORT_LIMIT, next_page=next_page) + if total is None: + total = resp.get("total", 0) + reports.extend(resp.get("items", [])) + if len(reports) >= total: + break + next_val = resp.get("next") + if not next_val: + break + next_page = int(next_val) + return reports + + @staticmethod + def create_report(central_conn, name, report_type, **kwargs): + """Create a new scheduled report. + + This method makes an API call to the following endpoint - + ``POST network-monitoring/v1alpha1/reports`` + + Args: + central_conn (NewCentralBase): Central connection object. + name (str): Report name. + report_type (str): Report type. + **kwargs: Additional report fields (schedule, filters, etc. per API spec). + + Returns: + (dict): API response with the created report. + + Raises: + ParameterError: If required fields missing. + """ + if not central_conn: + raise ParameterError("central_conn is required") + if not name or not isinstance(name, str): + raise ParameterError("name is required and must be a string") + if not report_type or not isinstance(report_type, str): + raise ParameterError("report_type is required and must be a string") + body = {"name": name, "type": report_type, **kwargs} + path = generate_url(REPORTS_ENDPOINT, "monitoring", "v1alpha1") + resp = central_conn.command(api_method="POST", api_path=path, api_data=body) + if resp["code"] not in (200, 201): + raise Exception(f"Error creating report: {resp['code']} - {resp['msg']}") + return resp["msg"] + + @staticmethod + def get_report(central_conn, report_id): + """Retrieve a specific report. + + This method makes an API call to the following endpoint - + ``GET network-monitoring/v1alpha1/reports/{report_id}`` + + Args: + central_conn (NewCentralBase): Central connection object. + report_id (str): Report identifier. + + Returns: + (dict): Report details. + + Raises: + ParameterError: If report_id missing. + """ + if not report_id or not isinstance(report_id, str): + raise ParameterError("report_id is required and must be a string") + return execute_get(central_conn, endpoint=f"{REPORTS_ENDPOINT}/{report_id}", version="v1alpha1") + + @staticmethod + def update_report(central_conn, report_id, **kwargs): + """Update a saved report. + + This method makes an API call to the following endpoint - + ``PUT network-monitoring/v1alpha1/reports/{report_id}`` + + Args: + central_conn (NewCentralBase): Central connection object. + report_id (str): Report identifier. + **kwargs: Fields to update. + + Returns: + (dict): API response. + + Raises: + ParameterError: If report_id missing. + """ + if not report_id or not isinstance(report_id, str): + raise ParameterError("report_id is required and must be a string") + path = generate_url(f"{REPORTS_ENDPOINT}/{report_id}", "monitoring", "v1alpha1") + resp = central_conn.command(api_method="PUT", api_path=path, api_data=kwargs) + if resp["code"] != 200: + raise Exception(f"Error updating report {report_id}: {resp['code']} - {resp['msg']}") + return resp["msg"] + + @staticmethod + def delete_report(central_conn, report_id): + """Delete a saved report. + + This method makes an API call to the following endpoint - + ``DELETE network-monitoring/v1alpha1/reports/{report_id}`` + + Args: + central_conn (NewCentralBase): Central connection object. + report_id (str): Report identifier. + + Returns: + (bool): True if deleted successfully. + + Raises: + ParameterError: If report_id missing. + """ + if not report_id or not isinstance(report_id, str): + raise ParameterError("report_id is required and must be a string") + path = generate_url(f"{REPORTS_ENDPOINT}/{report_id}", "monitoring", "v1alpha1") + resp = central_conn.command(api_method="DELETE", api_path=path) + if resp["code"] != 200: + raise Exception(f"Error deleting report {report_id}: {resp['code']} - {resp['msg']}") + return True + + @staticmethod + def list_report_runs(central_conn, report_id, limit=REPORT_LIMIT, next_page=1): + """List execution runs for a report. + + This method makes an API call to the following endpoint - + ``GET network-monitoring/v1alpha1/reports/{report_id}/runs`` + + Args: + central_conn (NewCentralBase): Central connection object. + report_id (str): Report identifier. + limit (int, optional): Max results per page (default 100). + next_page (int, optional): Pagination cursor (default 1). + + Returns: + (dict): API response with run history. + + Raises: + ParameterError: If report_id missing or limit/next_page invalid. + """ + if not report_id or not isinstance(report_id, str): + raise ParameterError("report_id is required and must be a string") + if limit > REPORT_LIMIT: + raise ParameterError(f"limit cannot exceed {REPORT_LIMIT}") + if next_page < 1: + raise ParameterError("next_page must be 1 or greater") + params = {"limit": limit, "next": next_page} + return execute_get(central_conn, endpoint=f"{REPORTS_ENDPOINT}/{report_id}/runs", + params=params, version="v1alpha1") diff --git a/pycentral/new_monitoring/switches.py b/pycentral/new_monitoring/switches.py new file mode 100644 index 0000000..92edcff --- /dev/null +++ b/pycentral/new_monitoring/switches.py @@ -0,0 +1,444 @@ +from ..utils.monitoring_utils import ( + execute_get, + generate_timestamp_str, + clean_raw_trend_data, + merged_dict_to_sorted_list, +) +from ..exceptions import ParameterError +from concurrent.futures import ThreadPoolExecutor, as_completed + +SWITCH_LIMIT = 100 +MONITOR_TYPE = "switches" + + +class MonitoringSwitches: + @staticmethod + def get_all_switches(central_conn, filter_str=None, sort=None): + """ + Retrieve all switches, handling pagination automatically. + + Args: + central_conn (NewCentralBase): Central connection object. + filter_str (str, optional): Optional filter expression (supported fields documented in API Reference Guide). + sort (str, optional): Optional sort parameter (supported fields documented in API Reference Guide). + + Returns: + (list[dict]): List of switch items. + """ + switches = [] + total_switches = None + next_page = 1 + while True: + resp = MonitoringSwitches.get_switches( + central_conn, + filter_str=filter_str, + sort=sort, + limit=SWITCH_LIMIT, + next_page=next_page, + ) + if total_switches is None: + total_switches = resp.get("total", 0) + + switches.extend(resp["items"]) + + if len(switches) == total_switches: + break + + next_page = resp.get("next") + if not next_page: + break + + next_page = int(next_page) + return switches + + @staticmethod + def get_switches( + central_conn, filter_str=None, sort=None, limit=SWITCH_LIMIT, next_page=1 + ): + """ + Retrieve a single page of switches. + + This method makes an API call to the following endpoint - `GET network-monitoring/v1/switches` + + Args: + central_conn (NewCentralBase): Central connection object. + filter_str (str, optional): Optional filter expression (supported fields documented in API Reference Guide). + sort (str, optional): Optional sort parameter (supported fields documented in API Reference Guide). + limit (int, optional): Number of entries to return (default is 100). + next_page (int, optional): Pagination cursor/index for next page (default is 1). + + Returns: + (dict): API response for the switches endpoint (contains 'items', 'total', etc.). + + Raises: + ParameterError: If limit or next_page values are invalid. + """ + path = MONITOR_TYPE + if limit > SWITCH_LIMIT: + raise ParameterError(f"limit cannot exceed {SWITCH_LIMIT}") + if next_page < 1: + raise ParameterError("next_page must be 1 or greater") + params = { + "filter": filter_str, + "sort": sort, + "limit": limit, + "next": next_page, + } + return execute_get(central_conn, endpoint=path, params=params) + + @staticmethod + def get_switch_details(central_conn, serial_number): + """ + Get details for a specific switch. + + This method makes an API call to the following endpoint - `GET network-monitoring/v1/switches/{serial-number}` + + Args: + central_conn (NewCentralBase): Central connection object. + serial_number (str): Serial number of the switch. + + Returns: + (dict): API response with switch details. + + Raises: + ParameterError: If serial_number is missing/invalid. + """ + MonitoringSwitches._validate_device_serial(serial_number=serial_number) + path = f"{MONITOR_TYPE}/{serial_number}" + return execute_get(central_conn, endpoint=path) + + @staticmethod + def get_switch_ports( + central_conn, + serial_number, + filter_str=None, + sort=None, + limit=100, + next_page=1, + ): + """ + Retrieve a list of ports for a specific switch. + + This method makes an API call to the following endpoint - `GET network-monitoring/v1/switches/{serial-number}/ports` + + Args: + central_conn (NewCentralBase): Central connection object. + serial_number (str): Serial number of the switch. + filter_str (str, optional): Optional filter expression (supported fields documented in API Reference Guide). + sort (str, optional): Optional sort parameter (supported fields documented in API Reference Guide). + limit (int, optional): Number of entries to return (default is 100). + next_page (int, optional): Pagination cursor/index for next page (default is 1). + + Returns: + (dict): API response for the switch ports endpoint (contains 'items', 'total', etc.). + + Raises: + ParameterError: If serial_number is missing/invalid or limit/next_page values are invalid. + """ + MonitoringSwitches._validate_device_serial(serial_number=serial_number) + if limit > 100: + raise ParameterError("limit cannot exceed 100") + if next_page < 1: + raise ParameterError("next_page must be 1 or greater") + path = f"{MONITOR_TYPE}/{serial_number}/ports" + params = { + "filter": filter_str, + "sort": sort, + "limit": limit, + "next": next_page, + } + return execute_get(central_conn, endpoint=path, params=params) + + @staticmethod + def get_switch_port_details(central_conn, serial_number, port_id): + """ + Get details for a specific port on a switch. + + This method makes an API call to the following endpoint - `GET network-monitoring/v1/switches/{serial-number}/ports/{port-id}` + + Args: + central_conn (NewCentralBase): Central connection object. + serial_number (str): Serial number of the switch. + port_id (str): Port identifier. + + Returns: + (dict): API response with port details. + + Raises: + ParameterError: If serial_number is missing/invalid or port_id is missing/invalid. + """ + MonitoringSwitches._validate_device_serial(serial_number=serial_number) + if not isinstance(port_id, str) or not port_id: + raise ParameterError("port_id is required and must be a string") + path = f"{MONITOR_TYPE}/{serial_number}/ports/{port_id}" + return execute_get(central_conn, endpoint=path) + + @staticmethod + def get_switch_cpu_utilization( + central_conn, + serial_number, + start_time=None, + end_time=None, + duration=None, + ): + """ + Retrieve CPU utilization trends for a switch. + + This method makes an API call to the following endpoint - `GET network-monitoring/v1/switches/{serial-number}/cpu-utilization-trends` + + Args: + central_conn (NewCentralBase): Central connection object. + serial_number (str): Serial number of the switch. + start_time (int, optional): Start time (epoch seconds) for range queries. + end_time (int, optional): End time (epoch seconds) for range queries. + duration (str|int, optional): Duration string or seconds for relative queries. + + Returns: + (dict|list): API response for cpu-utilization-trends. + + Raises: + ParameterError: If serial_number is missing/invalid. + """ + MonitoringSwitches._validate_device_serial(serial_number) + path = f"{MONITOR_TYPE}/{serial_number}/cpu-utilization-trends" + if start_time is None and end_time is None and duration is None: + return execute_get(central_conn, endpoint=path) + + return execute_get( + central_conn, + endpoint=path, + params={ + "filter": generate_timestamp_str( + start_time=start_time, end_time=end_time, duration=duration + ) + }, + ) + + @staticmethod + def get_switch_memory_utilization( + central_conn, + serial_number, + start_time=None, + end_time=None, + duration=None, + ): + """ + Retrieve memory utilization trends for a switch. + + This method makes an API call to the following endpoint - `GET network-monitoring/v1/switches/{serial-number}/memory-utilization-trends` + + Args: + central_conn (NewCentralBase): Central connection object. + serial_number (str): Serial number of the switch. + start_time (int, optional): Start time (epoch seconds) for range queries. + end_time (int, optional): End time (epoch seconds) for range queries. + duration (str|int, optional): Duration string or seconds for relative queries. + + Returns: + (dict|list): API response for memory-utilization-trends. + + Raises: + ParameterError: If serial_number is missing/invalid. + """ + MonitoringSwitches._validate_device_serial(serial_number) + path = f"{MONITOR_TYPE}/{serial_number}/memory-utilization-trends" + if start_time is None and end_time is None and duration is None: + return execute_get(central_conn, endpoint=path) + + return execute_get( + central_conn, + endpoint=path, + params={ + "filter": generate_timestamp_str( + start_time=start_time, end_time=end_time, duration=duration + ) + }, + ) + + @staticmethod + def get_switch_stats( + central_conn, + serial_number, + start_time=None, + end_time=None, + duration=None, + return_raw_response=False, + ): + """ + Collect multiple statistics (CPU, memory) for a switch for the specified time range. Default is to return sorted trend statistics for last 3 hours. + + Args: + central_conn (NewCentralBase): Central connection object. + serial_number (str): Serial number of the switch. + start_time (int, optional): Start time (epoch seconds) for range queries. + end_time (int, optional): End time (epoch seconds) for range queries. + duration (str|int, optional): Duration string (e.g. '5m') or seconds for relative queries. + return_raw_response (bool, optional): If True, return raw per-metric responses. + + Returns: + (list|dict): If return_raw_response is True returns raw list of responses; otherwise returns merged, sorted trend statistics for the switch. + + Raises: + ParameterError: If serial_number is missing/invalid. + RuntimeError: If any of the parallel metric requests fail. + """ + MonitoringSwitches._validate_device_serial(serial_number) + + # dispatch the two metric calls in parallel; helper methods handle timestamp logic + funcs = [ + MonitoringSwitches.get_switch_cpu_utilization, + MonitoringSwitches.get_switch_memory_utilization, + ] + + raw_results = [] + with ThreadPoolExecutor(max_workers=2) as executor: + future_map = { + executor.submit( + func, + central_conn, + serial_number, + start_time, + end_time, + duration, + ): func + for func in funcs + } + for fut in as_completed(future_map): + func = future_map[fut] + try: + resp = fut.result() + raw_results.append(resp) + except Exception as e: + # propagate the error for the caller to handle, but include which call failed + raise RuntimeError( + f"{func.__name__} metrics request failed: {e}" + ) from e + + if return_raw_response: + return raw_results + + data = {} + for resp in raw_results: + if not isinstance(resp, dict): + continue + data = clean_raw_trend_data(resp, data=data) + data = merged_dict_to_sorted_list(data) + return data + + @staticmethod + def get_switch_stacks( + central_conn, filter_str=None, sort=None, limit=100, next_page=1 + ): + """ + Retrieve a list of switch stacks. + + This method makes an API call to the following endpoint - `GET network-monitoring/v1/switch-stacks` + + Args: + central_conn (NewCentralBase): Central connection object. + filter_str (str, optional): Optional filter expression (supported fields documented in API Reference Guide). + sort (str, optional): Optional sort parameter (supported fields documented in API Reference Guide). + limit (int, optional): Number of entries to return (default is 100). + next_page (int, optional): Pagination cursor/index for next page (default is 1). + + Returns: + (dict): API response for the switch-stacks endpoint (contains 'items', 'total', etc.). + + Raises: + ParameterError: If limit or next_page values are invalid. + """ + path = "switch-stacks" + if limit > 100: + raise ParameterError("limit cannot exceed 100") + if next_page < 1: + raise ParameterError("next_page must be 1 or greater") + params = { + "filter": filter_str, + "sort": sort, + "limit": limit, + "next": next_page, + } + return execute_get(central_conn, endpoint=path, params=params) + + @staticmethod + def get_switch_stack_details(central_conn, stack_id): + """ + Get details for a specific switch stack. + + This method makes an API call to the following endpoint - `GET network-monitoring/v1/switch-stacks/{stack-id}` + + Args: + central_conn (NewCentralBase): Central connection object. + stack_id (str): Identifier of the switch stack. + + Returns: + (dict): API response with switch stack details. + + Raises: + ParameterError: If stack_id is missing/invalid. + """ + if not isinstance(stack_id, str) or not stack_id: + raise ParameterError("stack_id is required and must be a string") + path = f"switch-stacks/{stack_id}" + return execute_get(central_conn, endpoint=path) + + @staticmethod + def get_switch_vlans( + central_conn, + serial_number, + filter_str=None, + sort=None, + limit=100, + next_page=1, + ): + """ + Retrieve a list of VLANs for a specific switch. + + This method makes an API call to the following endpoint - `GET network-monitoring/v1/switches/{serial-number}/vlans` + + Args: + central_conn (NewCentralBase): Central connection object. + serial_number (str): Serial number of the switch. + filter_str (str, optional): Optional filter expression (supported fields documented in API Reference Guide). + sort (str, optional): Optional sort parameter (supported fields documented in API Reference Guide). + limit (int, optional): Number of entries to return (default is 100). + next_page (int, optional): Pagination cursor/index for next page (default is 1). + + Returns: + (dict): API response for the switch VLANs endpoint (contains 'items', 'total', etc.). + + Raises: + ParameterError: If serial_number is missing/invalid or limit/next_page values are invalid. + """ + MonitoringSwitches._validate_device_serial(serial_number=serial_number) + if limit > 100: + raise ParameterError("limit cannot exceed 100") + if next_page < 1: + raise ParameterError("next_page must be 1 or greater") + path = f"{MONITOR_TYPE}/{serial_number}/vlans" + params = { + "filter": filter_str, + "sort": sort, + "limit": limit, + "next": next_page, + } + return execute_get(central_conn, endpoint=path, params=params) + + @staticmethod + def _validate_device_serial(serial_number): + """ + Validate switch device serial_number. + + Args: + serial_number (str): Device serial number to validate. + + Raises: + ParameterError: If serial_number is missing or not a string. + + Note: + Internal SDK function + """ + if not isinstance(serial_number, str) or not serial_number: + raise ParameterError( + "serial_number is required and must be a string" + ) diff --git a/pycentral/scopes/__init__.py b/pycentral/scopes/__init__.py index 20204ec..bafb502 100644 --- a/pycentral/scopes/__init__.py +++ b/pycentral/scopes/__init__.py @@ -7,6 +7,7 @@ from .scope_maps import ScopeMaps from .device import Device from .device_group import Device_Group +from .device_group_api import DeviceGroupAPI __all__ = [ "Site", @@ -15,4 +16,5 @@ "ScopeMaps", "Device", "Device_Group", + "DeviceGroupAPI", ] diff --git a/pycentral/scopes/device_group_api.py b/pycentral/scopes/device_group_api.py new file mode 100644 index 0000000..de87d2e --- /dev/null +++ b/pycentral/scopes/device_group_api.py @@ -0,0 +1,324 @@ +# (C) Copyright 2025 Hewlett Packard Enterprise Development LP. +# MIT License + +from ..utils.url_utils import generate_url +from ..exceptions import ParameterError + +DEVICE_COLLECTION_ENDPOINT = "device-collections" +DEVICE_GROUP_LIMIT = 100 + + +class DeviceGroupAPI: + """Standalone API class for device group (device-collection) write operations.""" + + @staticmethod + def get_device_groups( + central_conn, filter_str=None, sort=None, limit=DEVICE_GROUP_LIMIT, next_page=1 + ): + """Retrieve a page of device groups. + + This method makes an API call to the following endpoint - + ``GET network-config/v1alpha1/device-collections`` + + Args: + central_conn (NewCentralBase): Central connection object. + filter_str (str, optional): OData filter expression. + sort (str, optional): Sort expression. + limit (int, optional): Number of entries to return per page (default 100). + next_page (int, optional): Pagination cursor (default 1). + + Returns: + (dict): API response body containing ``items``, ``total``, and ``next`` + fields. + + Raises: + ParameterError: If ``central_conn`` is None, ``limit`` exceeds + ``DEVICE_GROUP_LIMIT``, or ``next_page`` is less than 1. + """ + if not central_conn: + raise ParameterError("central_conn is required") + if limit > DEVICE_GROUP_LIMIT: + raise ParameterError(f"limit cannot exceed {DEVICE_GROUP_LIMIT}") + if next_page < 1: + raise ParameterError("next_page must be 1 or greater") + + params = { + "filter": filter_str, + "sort": sort, + "limit": limit, + "next": next_page, + } + + path = generate_url(DEVICE_COLLECTION_ENDPOINT, "configuration", "latest") + resp = central_conn.command("GET", path, api_params=params) + if resp["code"] != 200: + raise Exception( + f"Error retrieving device groups from {path}: " + f"{resp['code']} - {resp['msg']}" + ) + return resp["msg"] + + @staticmethod + def get_all_device_groups(central_conn, filter_str=None, sort=None): + """Retrieve all device groups, handling pagination automatically. + + Args: + central_conn (NewCentralBase): Central connection object. + filter_str (str, optional): OData filter expression. + sort (str, optional): Sort expression. + + Returns: + (list[dict]): All device group items across all pages. + + Raises: + ParameterError: If ``central_conn`` is None. + """ + items = [] + total = None + next_page = 1 + while True: + resp = DeviceGroupAPI.get_device_groups( + central_conn, + filter_str=filter_str, + sort=sort, + limit=DEVICE_GROUP_LIMIT, + next_page=next_page, + ) + if total is None: + total = resp.get("total", 0) + + items.extend(resp.get("items", [])) + + if total is not None and len(items) >= total: + break + + next_val = resp.get("next") + if not next_val: + break + next_page = int(next_val) + return items + + @staticmethod + def create_device_group(central_conn, name, description=None): + """Create a new device group. + + This method makes an API call to the following endpoint - + ``POST network-config/v1alpha1/device-collections`` + + Args: + central_conn (NewCentralBase): Central connection object. + name (str): Name of the device group. + description (str, optional): Description of the device group. + + Returns: + (dict): API response body. + + Raises: + ParameterError: If ``central_conn`` is None or ``name`` is missing. + Exception: If the API call fails. + """ + if not central_conn: + raise ParameterError("central_conn is required") + if not name: + raise ParameterError("name is required and cannot be empty") + + body = {"scopeName": name} + if description: + body["description"] = description + + path = generate_url(DEVICE_COLLECTION_ENDPOINT, "configuration", "latest") + resp = central_conn.command("POST", path, api_data=body) + if resp["code"] not in (200, 201): + raise Exception( + f"Error creating device group via {path}: " + f"{resp['code']} - {resp['msg']}" + ) + return resp["msg"] + + @staticmethod + def update_device_group(central_conn, scope_id, name=None, description=None): + """Update a device group. + + This method makes an API call to the following endpoint - + ``PUT network-config/v1alpha1/device-collections/{scope_id}`` + + Args: + central_conn (NewCentralBase): Central connection object. + scope_id (str | int): Scope ID of the device group to update. + name (str, optional): New name for the device group. + description (str, optional): New description for the device group. + + Returns: + (dict): API response body. + + Raises: + ParameterError: If ``central_conn`` or ``scope_id`` is missing, or if + neither ``name`` nor ``description`` is provided. + Exception: If the API call fails. + """ + if not central_conn: + raise ParameterError("central_conn is required") + if scope_id is None: + raise ParameterError("scope_id is required") + if name is None and description is None: + raise ParameterError( + "At least one of 'name' or 'description' must be provided" + ) + + body = {} + if name is not None: + body["scopeName"] = name + if description is not None: + body["description"] = description + + endpoint = f"{DEVICE_COLLECTION_ENDPOINT}/{scope_id}" + path = generate_url(endpoint, "configuration", "latest") + resp = central_conn.command("PUT", path, api_data=body) + if resp["code"] not in (200, 201): + raise Exception( + f"Error updating device group {scope_id} via {path}: " + f"{resp['code']} - {resp['msg']}" + ) + return resp["msg"] + + @staticmethod + def delete_device_group(central_conn, scope_id): + """Delete a device group by scope ID. + + This method makes an API call to the following endpoint - + ``DELETE network-config/v1alpha1/device-collections/{scope_id}`` + + Args: + central_conn (NewCentralBase): Central connection object. + scope_id (str | int): Scope ID of the device group to delete. + + Returns: + (dict): API response body. + + Raises: + ParameterError: If ``central_conn`` or ``scope_id`` is missing. + Exception: If the API call fails. + """ + if not central_conn: + raise ParameterError("central_conn is required") + if scope_id is None: + raise ParameterError("scope_id is required") + + endpoint = f"{DEVICE_COLLECTION_ENDPOINT}/{scope_id}" + path = generate_url(endpoint, "configuration", "latest") + resp = central_conn.command("DELETE", path) + if resp["code"] != 200: + raise Exception( + f"Error deleting device group {scope_id} via {path}: " + f"{resp['code']} - {resp['msg']}" + ) + return resp["msg"] + + @staticmethod + def delete_device_groups_bulk(central_conn, scope_ids): + """Bulk delete device groups. + + This method makes an API call to the following endpoint - + ``DELETE network-config/v1alpha1/device-collections`` + + Args: + central_conn (NewCentralBase): Central connection object. + scope_ids (list[str | int]): List of scope IDs to delete. + + Returns: + (dict): API response body. + + Raises: + ParameterError: If ``central_conn`` is None or ``scope_ids`` is empty. + Exception: If the API call fails. + """ + if not central_conn: + raise ParameterError("central_conn is required") + if not scope_ids: + raise ParameterError("scope_ids must be a non-empty list") + + params = {"scopeIds": [str(sid) for sid in scope_ids]} + path = generate_url(DEVICE_COLLECTION_ENDPOINT, "configuration", "latest") + resp = central_conn.command("DELETE", path, api_params=params) + if resp["code"] != 200: + raise Exception( + f"Error bulk-deleting device groups via {path}: " + f"{resp['code']} - {resp['msg']}" + ) + return resp["msg"] + + @staticmethod + def add_devices(central_conn, scope_id, serial_numbers): + """Add devices to a device group. + + This method makes an API call to the following endpoint - + ``POST network-config/v1alpha1/device-collections/{scope_id}/devices`` + + Args: + central_conn (NewCentralBase): Central connection object. + scope_id (str | int): Scope ID of the device group. + serial_numbers (list[str]): Serial numbers of devices to add. + + Returns: + (dict): API response body. + + Raises: + ParameterError: If ``central_conn`` or ``scope_id`` is missing, or if + ``serial_numbers`` is empty. + Exception: If the API call fails. + """ + if not central_conn: + raise ParameterError("central_conn is required") + if scope_id is None: + raise ParameterError("scope_id is required") + if not serial_numbers: + raise ParameterError("serial_numbers must be a non-empty list") + + body = {"serialNumbers": serial_numbers} + endpoint = f"{DEVICE_COLLECTION_ENDPOINT}/{scope_id}/devices" + path = generate_url(endpoint, "configuration", "latest") + resp = central_conn.command("POST", path, api_data=body) + if resp["code"] not in (200, 201): + raise Exception( + f"Error adding devices to device group {scope_id} via {path}: " + f"{resp['code']} - {resp['msg']}" + ) + return resp["msg"] + + @staticmethod + def remove_devices(central_conn, scope_id, serial_numbers): + """Remove devices from a device group. + + This method makes an API call to the following endpoint - + ``DELETE network-config/v1alpha1/device-collections/{scope_id}/devices`` + + Args: + central_conn (NewCentralBase): Central connection object. + scope_id (str | int): Scope ID of the device group. + serial_numbers (list[str]): Serial numbers of devices to remove. + + Returns: + (dict): API response body. + + Raises: + ParameterError: If ``central_conn`` or ``scope_id`` is missing, or if + ``serial_numbers`` is empty. + Exception: If the API call fails. + """ + if not central_conn: + raise ParameterError("central_conn is required") + if scope_id is None: + raise ParameterError("scope_id is required") + if not serial_numbers: + raise ParameterError("serial_numbers must be a non-empty list") + + params = {"serialNumbers": serial_numbers} + endpoint = f"{DEVICE_COLLECTION_ENDPOINT}/{scope_id}/devices" + path = generate_url(endpoint, "configuration", "latest") + resp = central_conn.command("DELETE", path, api_params=params) + if resp["code"] != 200: + raise Exception( + f"Error removing devices from device group {scope_id} via {path}: " + f"{resp['code']} - {resp['msg']}" + ) + return resp["msg"] diff --git a/pycentral/services/__init__.py b/pycentral/services/__init__.py new file mode 100644 index 0000000..7ecaad1 --- /dev/null +++ b/pycentral/services/__init__.py @@ -0,0 +1,6 @@ +# (C) Copyright 2025 Hewlett Packard Enterprise Development LP. +# MIT License + +from .webhooks import Webhooks +from .firmware import FirmwareService +from .audit import AuditTrail diff --git a/pycentral/services/audit.py b/pycentral/services/audit.py new file mode 100644 index 0000000..a3abfb0 --- /dev/null +++ b/pycentral/services/audit.py @@ -0,0 +1,166 @@ +# (C) Copyright 2025 Hewlett Packard Enterprise Development LP. +# MIT License + +from ..utils.url_utils import generate_url +from ..utils.monitoring_utils import build_timestamp_filter +from ..exceptions import ParameterError + +AUDIT_ENDPOINT = "audit" +AUDIT_LIMIT = 100 + + +class AuditTrail: + @staticmethod + def get_audit_events( + central_conn, + filter_str=None, + sort=None, + limit=100, + next_page=1, + start_time=None, + end_time=None, + duration=None, + ): + """Retrieve audit trail events. + + This method makes an API call to the following endpoint - `GET network-services/v1/audit` + + Args: + central_conn (NewCentralBase): Central connection object. + filter_str (str, optional): OData filter expression. + sort (str, optional): Sort expression. + limit (int, optional): Max results per page (default 100). + next_page (int, optional): Pagination cursor (default 1). + start_time (str, optional): Start time as RFC3339 string or Unix epoch + (ms or s). Must be provided together with end_time. + end_time (str, optional): End time as RFC3339 string or Unix epoch + (ms or s). Must be provided together with start_time. + duration (str, optional): Relative duration string (e.g. '3h', '1d', '1w'). + Cannot be combined with start_time/end_time. + + Returns: + (dict): API response containing audit event items, total count, and + pagination cursor. + + Raises: + ParameterError: If central_conn is None, limit exceeds 100, or next_page + is less than 1. + ValueError: If invalid time range parameters are provided. + """ + if not central_conn: + raise ParameterError("central_conn is required") + if limit > AUDIT_LIMIT: + raise ParameterError(f"limit cannot exceed {AUDIT_LIMIT}") + if next_page < 1: + raise ParameterError("next_page must be 1 or greater") + + params = { + "filter": filter_str, + "sort": sort, + "limit": limit, + "next": next_page, + } + + if start_time is not None or end_time is not None or duration is not None: + start_rfc, end_rfc = build_timestamp_filter( + start_time=start_time, + end_time=end_time, + duration=duration, + fmt="rfc3339", + ) + params["start-at"] = start_rfc + params["end-at"] = end_rfc + + path = generate_url(AUDIT_ENDPOINT, "services", "v1") + resp = central_conn.command("GET", path, api_params=params) + if resp["code"] != 200: + raise Exception( + f"Error retrieving audit events from {path}: {resp['code']} - {resp['msg']}" + ) + return resp["msg"] + + @staticmethod + def get_all_audit_events( + central_conn, + filter_str=None, + sort=None, + start_time=None, + end_time=None, + duration=None, + ): + """Retrieve all audit events, handling pagination automatically. + + Args: + central_conn (NewCentralBase): Central connection object. + filter_str (str, optional): OData filter expression. + sort (str, optional): Sort expression. + start_time (str, optional): Start time as RFC3339 string or Unix epoch + (ms or s). Must be provided together with end_time. + end_time (str, optional): End time as RFC3339 string or Unix epoch + (ms or s). Must be provided together with start_time. + duration (str, optional): Relative duration string (e.g. '3h', '1d', '1w'). + Cannot be combined with start_time/end_time. + + Returns: + (list): All audit event items across all pages. + + Raises: + ParameterError: If central_conn is None. + ValueError: If invalid time range parameters are provided. + """ + events = [] + total = None + next_page = 1 + while True: + resp = AuditTrail.get_audit_events( + central_conn, + filter_str=filter_str, + sort=sort, + limit=AUDIT_LIMIT, + next_page=next_page, + start_time=start_time, + end_time=end_time, + duration=duration, + ) + if total is None: + total = resp.get("total", 0) + + events.extend(resp.get("items", [])) + + if total is not None and len(events) >= total: + break + + next_val = resp.get("next") + if not next_val: + break + next_page = int(next_val) + return events + + @staticmethod + def get_audit_event_details(central_conn, event_id): + """Retrieve details for a specific audit event. + + This method makes an API call to the following endpoint - `GET network-services/v1/audit/{event_id}` + + Args: + central_conn (NewCentralBase): Central connection object. + event_id (str): Unique identifier of the audit event. + + Returns: + (dict): Audit event details as returned by the API. + + Raises: + ParameterError: If central_conn is None or event_id is missing or not a string. + """ + if not central_conn: + raise ParameterError("central_conn is required") + if not event_id or not isinstance(event_id, str): + raise ParameterError("event_id is required and must be a non-empty string") + + path = generate_url(f"{AUDIT_ENDPOINT}/{event_id}", "services", "v1") + resp = central_conn.command("GET", path, api_params={}) + if resp["code"] != 200: + raise Exception( + f"Error retrieving audit event {event_id}: {resp['code']} - {resp['msg']}" + ) + return resp["msg"] diff --git a/pycentral/services/firmware.py b/pycentral/services/firmware.py new file mode 100644 index 0000000..eafdcc1 --- /dev/null +++ b/pycentral/services/firmware.py @@ -0,0 +1,94 @@ +# (C) Copyright 2025 Hewlett Packard Enterprise Development LP. +# MIT License + +from ..utils.url_utils import generate_url +from ..exceptions import ParameterError + +FIRMWARE_DETAILS_ENDPOINT = "firmware-details" +FIRMWARE_LIMIT = 100 + + +class FirmwareService: + @staticmethod + def get_firmware_details(central_conn, filter_str=None, sort=None, limit=100, next_page=1): + """Retrieve firmware details list. + + This method makes an API call to the following endpoint - `GET network-services/v1/firmware-details` + + Args: + central_conn (NewCentralBase): Central connection object. + filter_str (str, optional): OData filter expression. + sort (str, optional): Sort expression. + limit (int, optional): Max results per page (default 100). + next_page (int, optional): Pagination cursor (default 1). + + Returns: + (dict): API response with firmware details items, including fields such as + firmware_version, upgrade_status, recommended_version, + firmware_classification, and last_upgraded_timestamp. + + Raises: + ParameterError: If central_conn is None, limit exceeds 100, or next_page is + less than 1. + """ + if not central_conn: + raise ParameterError("central_conn is required") + if limit > FIRMWARE_LIMIT: + raise ParameterError(f"limit cannot exceed {FIRMWARE_LIMIT}") + if next_page < 1: + raise ParameterError("next_page must be 1 or greater") + + params = { + "filter": filter_str, + "sort": sort, + "limit": limit, + "next": next_page, + } + + path = generate_url(FIRMWARE_DETAILS_ENDPOINT, "services", "v1") + resp = central_conn.command("GET", path, api_params=params) + if resp["code"] != 200: + raise Exception( + f"Error retrieving firmware details from {path}: {resp['code']} - {resp['msg']}" + ) + return resp["msg"] + + @staticmethod + def get_all_firmware_details(central_conn, filter_str=None, sort=None): + """Retrieve all firmware details, handling pagination automatically. + + Args: + central_conn (NewCentralBase): Central connection object. + filter_str (str, optional): OData filter expression. + sort (str, optional): Sort expression. + + Returns: + (list): All firmware detail items across all pages. + + Raises: + ParameterError: If central_conn is None. + """ + items = [] + total = None + next_page = 1 + while True: + resp = FirmwareService.get_firmware_details( + central_conn, + filter_str=filter_str, + sort=sort, + limit=FIRMWARE_LIMIT, + next_page=next_page, + ) + if total is None: + total = resp.get("total", 0) + + items.extend(resp.get("items", [])) + + if total is not None and len(items) >= total: + break + + next_val = resp.get("next") + if not next_val: + break + next_page = int(next_val) + return items diff --git a/pycentral/services/webhooks.py b/pycentral/services/webhooks.py new file mode 100644 index 0000000..0c18e5d --- /dev/null +++ b/pycentral/services/webhooks.py @@ -0,0 +1,237 @@ +# (C) Copyright 2025 Hewlett Packard Enterprise Development LP. +# MIT License + +from ..utils.url_utils import generate_url +from ..exceptions import ParameterError + +WEBHOOKS_ENDPOINT = "webhooks" + + +class Webhooks: + @staticmethod + def get_webhooks(central_conn): + """Retrieve a list of all webhooks. + + This method makes an API call to the following endpoint - `GET network-services/v1/webhooks` + + Args: + central_conn (NewCentralBase): Central connection object. + + Returns: + (list): List of webhook objects containing endpoint URL, auth mechanism, + createdAt, and updatedAt timestamps. + + Raises: + ParameterError: If central_conn is None. + """ + if not central_conn: + raise ParameterError("central_conn is required") + + path = generate_url(WEBHOOKS_ENDPOINT, "services", "v1") + resp = central_conn.command("GET", path, api_params={}) + if resp["code"] != 200: + raise Exception( + f"Error retrieving webhooks from {path}: {resp['code']} - {resp['msg']}" + ) + return resp["msg"] + + @staticmethod + def create_webhook(central_conn, name, url, auth_mechanism=None, subscriptions=None): + """Create a new webhook. + + This method makes an API call to the following endpoint - `POST network-services/v1/webhooks` + + Args: + central_conn (NewCentralBase): Central connection object. + name (str): Webhook name. + url (str): Receiver endpoint URL. + auth_mechanism (str, optional): Authentication mechanism for webhook delivery. + subscriptions (list, optional): List of event type strings to subscribe to. + + Returns: + (dict): Created webhook object. + + Raises: + ParameterError: If central_conn is None, name or url are missing or not strings. + """ + if not central_conn: + raise ParameterError("central_conn is required") + if not name or not isinstance(name, str): + raise ParameterError("name is required and must be a non-empty string") + if not url or not isinstance(url, str): + raise ParameterError("url is required and must be a non-empty string") + + body = {"name": name, "url": url} + if auth_mechanism is not None: + body["authMechanism"] = auth_mechanism + if subscriptions is not None: + body["subscriptions"] = subscriptions + + path = generate_url(WEBHOOKS_ENDPOINT, "services", "v1") + resp = central_conn.command(api_method="POST", api_path=path, api_data=body) + if resp["code"] not in (200, 201): + raise Exception( + f"Error creating webhook: {resp['code']} - {resp['msg']}" + ) + return resp["msg"] + + @staticmethod + def get_webhook(central_conn, webhook_id): + """Retrieve details for a specific webhook. + + This method makes an API call to the following endpoint - `GET network-services/v1/webhooks/{webhook_id}` + + Args: + central_conn (NewCentralBase): Central connection object. + webhook_id (str): Unique identifier of the webhook. + + Returns: + (dict): Webhook details. + + Raises: + ParameterError: If central_conn is None or webhook_id is missing or not a string. + """ + if not central_conn: + raise ParameterError("central_conn is required") + if not webhook_id or not isinstance(webhook_id, str): + raise ParameterError("webhook_id is required and must be a non-empty string") + + path = generate_url(f"{WEBHOOKS_ENDPOINT}/{webhook_id}", "services", "v1") + resp = central_conn.command("GET", path, api_params={}) + if resp["code"] != 200: + raise Exception( + f"Error retrieving webhook {webhook_id}: {resp['code']} - {resp['msg']}" + ) + return resp["msg"] + + @staticmethod + def update_webhook(central_conn, webhook_id, name=None, url=None, auth_mechanism=None, subscriptions=None): + """Update a webhook using a full replacement (PUT). + + This method makes an API call to the following endpoint - `PUT network-services/v1/webhooks/{webhook_id}` + + Args: + central_conn (NewCentralBase): Central connection object. + webhook_id (str): Unique identifier of the webhook to update. + name (str, optional): New webhook name. + url (str, optional): New receiver endpoint URL. + auth_mechanism (str, optional): New authentication mechanism. + subscriptions (list, optional): New list of event type strings to subscribe to. + + Returns: + (dict): Updated webhook object. + + Raises: + ParameterError: If central_conn is None or webhook_id is missing or not a string. + """ + if not central_conn: + raise ParameterError("central_conn is required") + if not webhook_id or not isinstance(webhook_id, str): + raise ParameterError("webhook_id is required and must be a non-empty string") + + body = {} + if name is not None: + body["name"] = name + if url is not None: + body["url"] = url + if auth_mechanism is not None: + body["authMechanism"] = auth_mechanism + if subscriptions is not None: + body["subscriptions"] = subscriptions + + path = generate_url(f"{WEBHOOKS_ENDPOINT}/{webhook_id}", "services", "v1") + resp = central_conn.command(api_method="PUT", api_path=path, api_data=body) + if resp["code"] != 200: + raise Exception( + f"Error updating webhook {webhook_id}: {resp['code']} - {resp['msg']}" + ) + return resp["msg"] + + @staticmethod + def patch_webhook(central_conn, webhook_id, **kwargs): + """Partially update a webhook (PATCH). + + This method makes an API call to the following endpoint - `PATCH network-services/v1/webhooks/{webhook_id}` + + Args: + central_conn (NewCentralBase): Central connection object. + webhook_id (str): Unique identifier of the webhook to patch. + **kwargs: Arbitrary keyword arguments representing fields to update + (e.g. name, url, authMechanism, subscriptions). + + Returns: + (dict): Updated webhook object. + + Raises: + ParameterError: If central_conn is None or webhook_id is missing or not a string. + """ + if not central_conn: + raise ParameterError("central_conn is required") + if not webhook_id or not isinstance(webhook_id, str): + raise ParameterError("webhook_id is required and must be a non-empty string") + + path = generate_url(f"{WEBHOOKS_ENDPOINT}/{webhook_id}", "services", "v1") + resp = central_conn.command(api_method="PATCH", api_path=path, api_data=kwargs) + if resp["code"] != 200: + raise Exception( + f"Error patching webhook {webhook_id}: {resp['code']} - {resp['msg']}" + ) + return resp["msg"] + + @staticmethod + def delete_webhook(central_conn, webhook_id): + """Delete a webhook. + + This method makes an API call to the following endpoint - `DELETE network-services/v1/webhooks/{webhook_id}` + + Args: + central_conn (NewCentralBase): Central connection object. + webhook_id (str): Unique identifier of the webhook to delete. + + Returns: + (dict): API response confirming deletion. + + Raises: + ParameterError: If central_conn is None or webhook_id is missing or not a string. + """ + if not central_conn: + raise ParameterError("central_conn is required") + if not webhook_id or not isinstance(webhook_id, str): + raise ParameterError("webhook_id is required and must be a non-empty string") + + path = generate_url(f"{WEBHOOKS_ENDPOINT}/{webhook_id}", "services", "v1") + resp = central_conn.command(api_method="DELETE", api_path=path) + if resp["code"] not in (200, 204): + raise Exception( + f"Error deleting webhook {webhook_id}: {resp['code']} - {resp['msg']}" + ) + return resp["msg"] + + @staticmethod + def rotate_hmac_key(central_conn, webhook_id): + """Rotate the HMAC key for a webhook. + + This method makes an API call to the following endpoint - `POST network-services/v1/webhooks/{webhook_id}/hmac-key` + + Args: + central_conn (NewCentralBase): Central connection object. + webhook_id (str): Unique identifier of the webhook whose HMAC key to rotate. + + Returns: + (dict): API response containing the new HMAC key details. + + Raises: + ParameterError: If central_conn is None or webhook_id is missing or not a string. + """ + if not central_conn: + raise ParameterError("central_conn is required") + if not webhook_id or not isinstance(webhook_id, str): + raise ParameterError("webhook_id is required and must be a non-empty string") + + path = generate_url(f"{WEBHOOKS_ENDPOINT}/{webhook_id}/hmac-key", "services", "v1") + resp = central_conn.command(api_method="POST", api_path=path, api_data={}) + if resp["code"] not in (200, 201): + raise Exception( + f"Error rotating HMAC key for webhook {webhook_id}: {resp['code']} - {resp['msg']}" + ) + return resp["msg"] diff --git a/pycentral/utils/url_utils.py b/pycentral/utils/url_utils.py index e411104..c90e823 100644 --- a/pycentral/utils/url_utils.py +++ b/pycentral/utils/url_utils.py @@ -1,7 +1,7 @@ # (C) Copyright 2025 Hewlett Packard Enterprise Development LP. # MIT License -versions = ["v1alpha1", "v1"] +versions = ["v1alpha1", "v1", "v1alpha2"] CATEGORIES = { "configuration": { @@ -27,6 +27,16 @@ "type": "glp", "latest": "v1", }, + "notifications": { + "value": "network-notifications", + "type": "monitoring", + "latest": "v1", + }, + "services": { + "value": "network-services", + "type": "monitoring", + "latest": "v1", + }, }