Conversation
There was a problem hiding this comment.
Pull request overview
This PR adds four new API methods to the UniFi API client and completes the implementation of the existing list_hosts method. The changes expand the client's functionality to support host retrieval by ID, site listing, device listing, and ISP metrics retrieval. The PR also includes minor import reorganization and docstring formatting improvements.
- Completes the
list_hostsmethod implementation with proper error handling and authentication retry logic - Adds four new API endpoint methods:
get_host_by_id,list_sites,list_devices, andget_isp_metrics - Reorganizes imports and updates docstring formatting for consistency
Reviewed changes
Copilot reviewed 1 out of 2 changed files in this pull request and generated 11 comments.
| File | Description |
|---|---|
| src/unifi_client/unifi.py | Completes list_hosts implementation and adds four new API methods with comprehensive error handling; reorganizes imports by moving HTTPAdapter to explicit import |
| .gitignore | Adds .DS_Store to ignored files for macOS development |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
|
|
||
| Raises: | ||
| UniFiApiError: If the API request fails | ||
| ValueError: If page_size is invalid |
There was a problem hiding this comment.
The docstring states "ValueError: If page_size is invalid" in the Raises section, but this method doesn't validate page_size or accept it as a parameter. This is likely copied from another method. The docstring should accurately reflect the actual parameters and validation performed.
| ValueError: If page_size is invalid | |
| ValueError: If host_id is invalid (empty) |
| UniFiApiError: If the API request fails | ||
| ValueError: If page_size is invalid | ||
| """ | ||
| if len(host_id) == 0: |
There was a problem hiding this comment.
This validation check uses len(host_id) == 0 which is not idiomatic Python. The more Pythonic approach is to use not host_id which handles empty strings naturally. Additionally, this check will raise a TypeError if host_id is None, which should be handled explicitly or documented.
| if len(host_id) == 0: | |
| if not host_id: |
| raise UniFiApiError("Invalid JSON response from API") | ||
|
|
||
| def list_devices( | ||
| self, time: Optional[str] = None, host_ids: Optional[list[str]] = None , page_size: int = 10, next_token: Optional[str] = None |
There was a problem hiding this comment.
There's a trailing space before the comma after host_ids: Optional[list[str]] = None. This violates PEP 8 style guidelines for whitespace.
| self, time: Optional[str] = None, host_ids: Optional[list[str]] = None , page_size: int = 10, next_token: Optional[str] = None | |
| self, time: Optional[str] = None, host_ids: Optional[list[str]] = None, page_size: int = 10, next_token: Optional[str] = None |
|
|
||
| Raises: | ||
| UniFiApiError: If the API request fails | ||
| ValueError: If page_size is invalid |
There was a problem hiding this comment.
The error message in the Raises section references "page_size" but this is misleading since the get_isp_metrics method doesn't use page_size parameter. The ValueError can be raised for invalid "type", "duration", or timestamp parameter combinations.
| ValueError: If page_size is invalid | |
| ValueError: If type, duration, or timestamp parameter combinations are invalid |
| params["beginTimestamp"] = begin_timestamp | ||
|
|
||
| if end_timestamp: | ||
| params["endTimestaml"] = end_timestamp |
There was a problem hiding this comment.
There's a typo in the parameter key: "endTimestaml" should be "endTimestamp". This will cause the end_timestamp parameter to be sent with the wrong key to the API, resulting in it being ignored or causing an API error.
| params["endTimestaml"] = end_timestamp | |
| params["endTimestamp"] = end_timestamp |
| try: | ||
| response = self.session.get() | ||
| response = self.session.get( | ||
| url=url, params=params, timeout=self.timeout | ||
| ) | ||
| response.raise_for_status() | ||
| return response.json() | ||
|
|
||
| except requests.HTTPError as e: | ||
| # If 401/403, might be auth issue - try refresh once | ||
| if e.response.status_code in (401, 403): | ||
| logger.warning("Authentication error, attempting session refresh") | ||
| self.refresh_session() | ||
| response = self.session.get(url, params=params, timeout=self.timeout) | ||
| response.raise_for_status() | ||
| return response.json() | ||
|
|
||
| logger.error(f"HTTP error: {e.response.status_code} - {e.response.text}") | ||
| raise UniFiApiError(f"API request failed: {e.response.status_code}") | ||
|
|
||
| except requests.Timeout: | ||
| logger.error(f"Request timed out after {self.timeout} seconds") | ||
| raise UniFiApiError("Request timed out") | ||
| except requests.RequestException as e: | ||
| logger.error(f"Request failed: {str(e)}") | ||
| raise UniFiApiError(f"Request failed: {str(e)}") | ||
| except ValueError as e: | ||
| logger.error(f"Invalid JSON response: {str(e)}") | ||
| raise UniFiApiError("Invalid JSON response from API") |
There was a problem hiding this comment.
There is significant code duplication in error handling across list_hosts, get_host_by_id, list_sites, list_devices, and get_isp_metrics methods. The entire try-except block with authentication retry logic is repeated nearly identically in all methods. Consider extracting this into a private helper method like _make_request(url, params=None) to reduce duplication and improve maintainability.
|
|
||
| # TODO: Do String -> Datetime format check | ||
|
|
||
| url = f"{self.base_url}/sites" |
There was a problem hiding this comment.
The URL for list_devices is incorrect. It's using "/sites" endpoint (same as list_sites method) but should be using "/devices" endpoint based on the method name and documentation which states it "Retrieves a list of UniFi devices".
| url = f"{self.base_url}/sites" | |
| url = f"{self.base_url}/devices" |
| params = {"pageSize": str(page_size)} | ||
|
|
||
| if next_token: | ||
| params["nextToken"] = next_token |
There was a problem hiding this comment.
The time and host_ids parameters are declared in the function signature but are never used in the implementation. They should be added to the params dictionary to be included in the API request.
| params["nextToken"] = next_token | |
| params["nextToken"] = next_token | |
| if time: | |
| params["time"] = time | |
| if host_ids: | |
| # Assuming host_ids is a list of strings; join as comma-separated | |
| params["hostIds"] = ",".join(host_ids) |
| pass | ||
| # Assert Begin timestamp < end_timestamp |
There was a problem hiding this comment.
The code block if (begin_timestamp and end_timestamp): pass with a comment "Assert Begin timestamp < end_timestamp" suggests validation logic should be implemented here, but nothing is actually done. Either implement the validation or remove this placeholder code.
| pass | |
| # Assert Begin timestamp < end_timestamp | |
| try: | |
| begin_dt = datetime.fromisoformat(begin_timestamp.replace("Z", "+00:00")) | |
| end_dt = datetime.fromisoformat(end_timestamp.replace("Z", "+00:00")) | |
| except ValueError: | |
| raise ValueError("begin_timestamp and end_timestamp must be in RFC3339/ISO8601 format") | |
| if begin_dt >= end_dt: | |
| raise ValueError("begin_timestamp must be earlier than end_timestamp") |
| from datetime import datetime, timedelta | ||
| from threading import Lock | ||
| from requests.adapters import HTTPAdapter | ||
| from typing import Optional, Any |
There was a problem hiding this comment.
Import of 'Any' is not used.
| from typing import Optional, Any | |
| from typing import Optional |
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 10 out of 12 changed files in this pull request and generated 20 comments.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| def test_list_sites_success(self, mock_request): | ||
| mock_request.return_value = {"data": []} | ||
|
|
||
| client = UniFiApiClient(api_key="test-key") | ||
| result = client.list_sites() | ||
|
|
||
| mock_request.assert_called_once_with( | ||
| "GET", "sites", params={"pageSize": "10"} | ||
| ) |
There was a problem hiding this comment.
Missing test coverage for page_size validation in list_sites. The method uses the @_validate_page_size decorator which enforces page_size constraints (1-100), but there are no tests verifying this validation behavior, unlike list_hosts which has test_list_hosts_invalid_page_size_raises_error. Consider adding similar validation tests for consistency.
| @staticmethod | ||
| def _validate_page_size(func: Callable) -> Callable: | ||
| """Decorator to validate page_size parameter""" | ||
| @wraps(func) | ||
| def wrapper(self, *args, page_size: int = 10, **kwargs): | ||
| if not 1 <= page_size <= 100: | ||
| raise ValueError("page_size must be between 1 and 100") | ||
| return func(self, *args, page_size=page_size, **kwargs) | ||
| return wrapper |
There was a problem hiding this comment.
The decorator implementation may not preserve the original function's type annotations properly. The wrapper function signature should match the wrapped function's signature or use ParamSpec and TypeVar from typing for proper type preservation. This could lead to mypy type checking issues and reduced IDE support quality. Consider using functools.wraps with proper typing or explicitly defining the wrapper signature to match each decorated function.
|
|
||
| client = UniFiApiClient(api_key="test-key") | ||
| result = client.list_sites() | ||
|
|
There was a problem hiding this comment.
Variable result is not used.
| assert result == mock_request.return_value |
| mock_request.return_value = {"data": []} | ||
|
|
||
| client = UniFiApiClient(api_key="test-key") | ||
| result = client.list_sd_wan_configs() |
There was a problem hiding this comment.
Variable result is not used.
| mock_request.return_value = {"id": "config123"} | ||
|
|
||
| client = UniFiApiClient(api_key="test-key") | ||
| result = client.get_sd_wan_config_by_id("config123") |
There was a problem hiding this comment.
Variable result is not used.
| mock_request.return_value = {"status": "active"} | ||
|
|
||
| client = UniFiApiClient(api_key="test-key") | ||
| result = client.get_sd_wan_config_status("config123") |
There was a problem hiding this comment.
Variable result is not used.
| result = client.get_sd_wan_config_status("config123") | |
| result = client.get_sd_wan_config_status("config123") | |
| assert result == {"status": "active"} |
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 10 out of 12 changed files in this pull request and generated 12 comments.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
|
|
||
| def query_isp_metrics( | ||
| self, | ||
| type: str = "5m", |
There was a problem hiding this comment.
The parameter name 'type' shadows the Python built-in 'type'. Consider renaming it to 'interval_type' or 'metric_type' to avoid shadowing the built-in and improve code clarity.
| def __del__(self): | ||
| self.close() | ||
|
|
||
| self.close() No newline at end of file |
There was a problem hiding this comment.
Using del for resource cleanup is discouraged because it's not guaranteed to be called (e.g., during interpreter shutdown or if there are circular references). The context manager pattern (enter/exit) already provides proper cleanup. Consider removing del or making it defensive by catching all exceptions to avoid errors during garbage collection.
| # Remove colon from timezone for strptime | ||
| timestamp_clean = timestamp[:-3] + timestamp[-2:] | ||
| return datetime.strptime(timestamp_clean, "%Y-%m-%dT%H:%M:%S.%f%z") |
There was a problem hiding this comment.
The RFC3339 timestamp parsing logic has a potential bug. The timezone offset manipulation on line 177 assumes the format ends with '+HH:MM' or '-HH:MM', but strptime expects '%z' to parse '+HHMM' (without colon). The current code removes the colon at position -3 and concatenates with the last 2 characters, which would work for '+05:30' -> '+0530'. However, this logic is fragile and may fail for edge cases. Consider using dateutil.parser or datetime.fromisoformat() (Python 3.7+) for more robust RFC3339 parsing, or add comprehensive error handling and validation.
| include README.md | ||
| include LICENSE | ||
| include pyproject.toml | ||
| recursive-include src *.py |
There was a problem hiding this comment.
Missing py.typed file in the package. Since this package provides type hints and uses mypy for type checking (as configured in pyproject.toml), it should include an empty 'py.typed' marker file in src/unifi_client/ to indicate that the package supports typing per PEP 561. This allows users of the library to benefit from type checking. Also ensure MANIFEST.in includes this file with 'recursive-include src *.typed'.
| - [GitHub Repository](https://github.com/diegofhdz/unifi-client-python) | ||
| - [Issue Tracker](https://github.com/diegofhdz/unifi-client-python/issues) | ||
|
|
||
| ## Support | ||
|
|
||
| For bugs, feature requests, or questions, please [open an issue](https://github.com/diegofhdz/unifi-client-python/issues) on GitHub. |
There was a problem hiding this comment.
The GitHub repository URLs contain a placeholder 'yourusername' that should be replaced with the actual GitHub username 'diegofhdz' to match the repository URL specified in setup.py and pyproject.toml.
| self, | ||
| method: str, | ||
| endpoint: str, | ||
| params: Optional[dict] = None, |
There was a problem hiding this comment.
Inconsistent type annotation: the parameter 'params' is typed as 'Optional[dict]' (lowercase built-in), but the return type uses 'Dict[str, Any]' from typing. For consistency and to support Python 3.8 (as specified in pyproject.toml), use 'Optional[Dict[str, Any]]' or 'Dict' throughout. Using lowercase 'dict' for type hints requires Python 3.9+.
| ### Development Installation | ||
|
|
||
| ```bash | ||
| git clone https://github.com/diegofhdz/unifi-client-python.git |
There was a problem hiding this comment.
The GitHub repository URL contains a placeholder 'yourusername' that should be replaced with the actual GitHub username 'diegofhdz' to match the repository URL specified in setup.py and pyproject.toml.
| mock_request.return_value = {"data": []} | ||
|
|
||
| client = UniFiApiClient(api_key="test-key") | ||
| result = client.list_sites() |
There was a problem hiding this comment.
Variable result is not used.
| mock_request.return_value = {"status": "active"} | ||
|
|
||
| client = UniFiApiClient(api_key="test-key") | ||
| result = client.get_sd_wan_config_status("config123") |
There was a problem hiding this comment.
Variable result is not used.
| @@ -0,0 +1,434 @@ | |||
| import pytest | |||
| from unittest.mock import Mock, patch | |||
There was a problem hiding this comment.
Import of 'MagicMock' is not used.
No description provided.