diff --git a/pyoverkiz/_case.py b/pyoverkiz/_case.py index f87448b7..2170989f 100644 --- a/pyoverkiz/_case.py +++ b/pyoverkiz/_case.py @@ -4,8 +4,10 @@ import functools import re -from collections.abc import Callable -from typing import Any +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from collections.abc import Callable _CAMEL_RE = re.compile(r"([A-Z]+)([A-Z][a-z])|([a-z\d])([A-Z])") diff --git a/pyoverkiz/action_queue.py b/pyoverkiz/action_queue.py index 104809a8..2b5e3523 100644 --- a/pyoverkiz/action_queue.py +++ b/pyoverkiz/action_queue.py @@ -4,13 +4,14 @@ import asyncio import contextlib -from collections.abc import Callable, Coroutine, Generator from dataclasses import dataclass from typing import TYPE_CHECKING, Any from pyoverkiz.models import Action if TYPE_CHECKING: + from collections.abc import Callable, Coroutine, Generator + from pyoverkiz.enums import ExecutionMode diff --git a/pyoverkiz/auth/base.py b/pyoverkiz/auth/base.py index e10b99ca..bdf2a490 100644 --- a/pyoverkiz/auth/base.py +++ b/pyoverkiz/auth/base.py @@ -3,9 +3,11 @@ from __future__ import annotations import datetime -from collections.abc import Mapping from dataclasses import dataclass -from typing import Any, Protocol +from typing import TYPE_CHECKING, Any, Protocol + +if TYPE_CHECKING: + from collections.abc import Mapping @dataclass(slots=True) diff --git a/pyoverkiz/auth/factory.py b/pyoverkiz/auth/factory.py index 4ee91425..0e677e7b 100644 --- a/pyoverkiz/auth/factory.py +++ b/pyoverkiz/auth/factory.py @@ -2,9 +2,7 @@ from __future__ import annotations -import ssl - -from aiohttp import ClientSession +from typing import TYPE_CHECKING from pyoverkiz.auth.credentials import ( Credentials, @@ -24,7 +22,13 @@ SomfyAuthStrategy, ) from pyoverkiz.enums import APIType, Server -from pyoverkiz.models import ServerConfig + +if TYPE_CHECKING: + import ssl + + from aiohttp import ClientSession + + from pyoverkiz.models import ServerConfig def build_auth_strategy( diff --git a/pyoverkiz/auth/strategies.py b/pyoverkiz/auth/strategies.py index b5a0e3c7..0d44203c 100644 --- a/pyoverkiz/auth/strategies.py +++ b/pyoverkiz/auth/strategies.py @@ -6,23 +6,26 @@ import base64 import binascii import json -import ssl -from collections.abc import Mapping from http import HTTPStatus from typing import TYPE_CHECKING, Any, cast if TYPE_CHECKING: + import ssl + from collections.abc import Mapping + from botocore.client import BaseClient + from pyoverkiz.auth.credentials import ( + LocalTokenCredentials, + RexelOAuthCodeCredentials, + TokenCredentials, + UsernamePasswordCredentials, + ) + from pyoverkiz.models import ServerConfig + from aiohttp import ClientSession, FormData from pyoverkiz.auth.base import AuthContext, AuthStrategy -from pyoverkiz.auth.credentials import ( - LocalTokenCredentials, - RexelOAuthCodeCredentials, - TokenCredentials, - UsernamePasswordCredentials, -) from pyoverkiz.const import ( COZYTOUCH_ATLANTIC_API, COZYTOUCH_CLIENT_ID, @@ -48,7 +51,6 @@ SomfyBadCredentialsError, SomfyServiceError, ) -from pyoverkiz.models import ServerConfig MIN_JWT_SEGMENTS = 2 @@ -159,7 +161,7 @@ async def refresh_if_needed(self) -> bool: await self._request_access_token( grant_type="refresh_token", - extra_fields={"refresh_token": cast(str, self.context.refresh_token)}, + extra_fields={"refresh_token": cast("str", self.context.refresh_token)}, ) return True @@ -351,7 +353,7 @@ async def refresh_if_needed(self) -> bool: "grant_type": "refresh_token", "client_id": REXEL_OAUTH_CLIENT_ID, "scope": REXEL_OAUTH_SCOPE, - "refresh_token": cast(str, self.context.refresh_token), + "refresh_token": cast("str", self.context.refresh_token), } ) return True @@ -429,6 +431,6 @@ def _decode_jwt_payload(token: str) -> dict[str, Any]: padding = "=" * (-len(payload_segment) % 4) try: decoded = base64.urlsafe_b64decode(payload_segment + padding) - return cast(dict[str, Any], json.loads(decoded)) + return cast("dict[str, Any]", json.loads(decoded)) except (binascii.Error, json.JSONDecodeError) as error: raise InvalidTokenError("Malformed JWT received.") from error diff --git a/pyoverkiz/client.py b/pyoverkiz/client.py index 38d8dbba..c5b597d4 100644 --- a/pyoverkiz/client.py +++ b/pyoverkiz/client.py @@ -7,8 +7,7 @@ import urllib.parse from http import HTTPStatus from pathlib import Path -from types import TracebackType -from typing import Any, Self, cast +from typing import TYPE_CHECKING, Any, Self, cast import backoff from aiohttp import ( @@ -17,7 +16,6 @@ ClientSession, ServerDisconnectedError, ) -from backoff.types import Details from pyoverkiz._case import decamelize from pyoverkiz.action_queue import ActionQueue, ActionQueueSettings @@ -55,14 +53,20 @@ from pyoverkiz.obfuscate import obfuscate_sensitive_data from pyoverkiz.response_handler import check_response from pyoverkiz.serializers import prepare_payload -from pyoverkiz.types import JSON + +if TYPE_CHECKING: + from types import TracebackType + + from backoff.types import Details + + from pyoverkiz.types import JSON _LOGGER = logging.getLogger(__name__) def _get_client_from_invocation(invocation: Details) -> OverkizClient: """Return the `OverkizClient` instance from a backoff invocation.""" - return cast(OverkizClient, invocation["args"][0]) + return cast("OverkizClient", invocation["args"][0]) async def relogin(invocation: Details) -> None: @@ -419,7 +423,7 @@ async def register_event_listener(self) -> str: API on a regular basis. """ response = await self._post("events/register") - listener_id = cast(str, response.get("id")) + listener_id = cast("str", response.get("id")) self.event_listener_id = listener_id return listener_id @@ -472,7 +476,7 @@ async def get_api_version(self) -> str: """Get the API version (local only).""" response = await self._get("apiVersion") - return cast(str, response["protocolVersion"]) + return cast("str", response["protocolVersion"]) @retry_on_too_many_executions @retry_on_auth_error @@ -492,7 +496,7 @@ async def _execute_action_group_direct( response: dict = await self._post(url, prepare_payload(payload)) - return cast(str, response["execId"]) + return cast("str", response["execId"]) async def execute_action_group( self, @@ -578,13 +582,13 @@ async def get_places(self) -> Place: async def execute_persisted_action_group(self, oid: str) -> str: """Execute a server-side action group by its OID (see ``get_action_groups``).""" response = await self._post(f"exec/{oid}") - return cast(str, response["execId"]) + return cast("str", response["execId"]) @retry_on_auth_error async def schedule_persisted_action_group(self, oid: str, timestamp: int) -> str: """Schedule a server-side action group for execution at the given timestamp.""" response = await self._post(f"exec/schedule/{oid}/{timestamp}") - return cast(str, response["triggerId"]) + return cast("str", response["triggerId"]) @retry_on_auth_error async def get_setup_options(self) -> list[Option]: diff --git a/pyoverkiz/enums/base.py b/pyoverkiz/enums/base.py index 83c075bc..8ce8dd35 100644 --- a/pyoverkiz/enums/base.py +++ b/pyoverkiz/enums/base.py @@ -36,4 +36,4 @@ def _missing_(cls, value: object) -> Self: # type: ignore[override] message = cls.__missing_message__ logging.getLogger(cls.__module__).warning(message, value, cls) # Type checker cannot infer UNKNOWN exists on Self, but all subclasses define it - return cast(Self, cls.UNKNOWN) # type: ignore[attr-defined] + return cast("Self", cls.UNKNOWN) # type: ignore[attr-defined] diff --git a/pyoverkiz/models.py b/pyoverkiz/models.py index 411ec73d..6faab4ac 100644 --- a/pyoverkiz/models.py +++ b/pyoverkiz/models.py @@ -5,8 +5,7 @@ import json import logging import re -from collections.abc import Iterator -from typing import Any, cast +from typing import TYPE_CHECKING, Any, cast from attr import define, field @@ -24,12 +23,16 @@ UIWidget, UpdateBoxStatus, ) -from pyoverkiz.enums.command import OverkizCommand, OverkizCommandParam from pyoverkiz.enums.protocol import Protocol from pyoverkiz.enums.server import APIType, Server from pyoverkiz.obfuscate import obfuscate_email, obfuscate_id, obfuscate_string from pyoverkiz.types import DATA_TYPE_TO_PYTHON, StateType +if TYPE_CHECKING: + from collections.abc import Iterator + + from pyoverkiz.enums.command import OverkizCommand, OverkizCommandParam + # pylint: disable=unused-argument, too-many-instance-attributes, too-many-locals # :///[#] @@ -527,7 +530,7 @@ def value_as_int(self) -> int | None: if self.type == DataType.NONE: return None if self.type == DataType.INTEGER: - return cast(int, self.value) + return cast("int", self.value) raise TypeError(f"{self.name} is not an integer") @property @@ -536,9 +539,9 @@ def value_as_float(self) -> float | None: if self.type == DataType.NONE: return None if self.type == DataType.FLOAT: - return cast(float, self.value) + return cast("float", self.value) if self.type == DataType.INTEGER: - return float(cast(int, self.value)) + return float(cast("int", self.value)) raise TypeError(f"{self.name} is not a float") @property @@ -547,7 +550,7 @@ def value_as_bool(self) -> bool | None: if self.type == DataType.NONE: return None if self.type == DataType.BOOLEAN: - return cast(bool, self.value) + return cast("bool", self.value) raise TypeError(f"{self.name} is not a boolean") @property @@ -556,7 +559,7 @@ def value_as_str(self) -> str | None: if self.type == DataType.NONE: return None if self.type == DataType.STRING: - return cast(str, self.value) + return cast("str", self.value) raise TypeError(f"{self.name} is not a string") @property @@ -565,7 +568,7 @@ def value_as_dict(self) -> dict[str, Any] | None: if self.type == DataType.NONE: return None if self.type == DataType.JSON_OBJECT: - return cast(dict, self.value) + return cast("dict", self.value) raise TypeError(f"{self.name} is not a JSON object") @property @@ -574,7 +577,7 @@ def value_as_list(self) -> list[Any] | None: if self.type == DataType.NONE: return None if self.type == DataType.JSON_ARRAY: - return cast(list, self.value) + return cast("list", self.value) raise TypeError(f"{self.name} is not an array") @@ -899,7 +902,10 @@ def __init__( **_: Any, ) -> None: """Initialize ActionGroup from API data and convert nested actions.""" - self.id = oid or id + if oid is None and id is None: + raise ValueError("Either 'oid' or 'id' must be provided") + + self.id = cast("str", oid or id) self.creation_time = creation_time self.last_update_time = last_update_time self.label = ( @@ -912,7 +918,7 @@ def __init__( self.notification_text = notification_text self.notification_title = notification_title self.actions = [Action(**action) for action in actions] - self.oid = oid or id + self.oid = cast("str", oid or id) @define(init=False, kw_only=True) @@ -1421,7 +1427,12 @@ def __init__( @define(kw_only=True) class FirmwareStatus: - """Firmware status of a device.""" + """ + Firmware status of a device. + + `up_to_date` indicates whether the current firmware is the latest available + version and `notes` contains provider messages about the firmware state. + """ up_to_date: bool notes: list[dict[str, str]] diff --git a/pyoverkiz/obfuscate.py b/pyoverkiz/obfuscate.py index ca26abd6..0831bad8 100644 --- a/pyoverkiz/obfuscate.py +++ b/pyoverkiz/obfuscate.py @@ -3,9 +3,10 @@ from __future__ import annotations import re -from typing import Any +from typing import TYPE_CHECKING, Any -from pyoverkiz.types import JSON +if TYPE_CHECKING: + from pyoverkiz.types import JSON def obfuscate_id(id: str | None) -> str: diff --git a/pyoverkiz/response_handler.py b/pyoverkiz/response_handler.py index dbeb918c..66ea0952 100644 --- a/pyoverkiz/response_handler.py +++ b/pyoverkiz/response_handler.py @@ -4,8 +4,7 @@ from http import HTTPStatus from json import JSONDecodeError - -from aiohttp import ClientResponse +from typing import TYPE_CHECKING from pyoverkiz.exceptions import ( AccessDeniedToGatewayError, @@ -40,6 +39,9 @@ UnsupportedOperationError, ) +if TYPE_CHECKING: + from aiohttp import ClientResponse + # Primary dispatch: (errorCode, message_substring) -> error class. # Checked in order; first match wins. Use errorCode as the primary key to # reduce brittleness across cloud vs. local API variants. diff --git a/pyoverkiz/types.py b/pyoverkiz/types.py index 167f5951..fdd47498 100644 --- a/pyoverkiz/types.py +++ b/pyoverkiz/types.py @@ -3,11 +3,13 @@ from __future__ import annotations import json -from collections.abc import Callable -from typing import Any +from typing import TYPE_CHECKING, Any from pyoverkiz.enums import DataType +if TYPE_CHECKING: + from collections.abc import Callable + StateType = str | int | float | bool | dict[str, Any] | list[Any] | None diff --git a/pyproject.toml b/pyproject.toml index f97104c5..d3478598 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -124,6 +124,8 @@ select = [ "A", # pylint "PL", + # flake8-type-checking + "TC", ] ignore = [ "E501", # Line too long @@ -132,6 +134,7 @@ ignore = [ "PLR0913", # Too many arguments — expected for API model constructors "PLC0415", # Import not at top level — intentional lazy imports "PLR0911", # Too many return statements — acceptable in factory functions + "TC006", # Quoting cast() types hurts IDE autocomplete for no real benefit ] [tool.ruff.lint.per-file-ignores] diff --git a/tests/test_client.py b/tests/test_client.py index 17bbbb9b..97084580 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -725,7 +725,7 @@ async def test_get_current_execution_returns_none_for_empty_response( client: OverkizClient, fixture_name: str, ): - """Cloud returns {} and local returns [] for non-existent exec_ids.""" + """Cloud returns {} and local returns [] for non-existent exec_id.""" with (CURRENT_DIR / "fixtures" / "endpoints" / fixture_name).open( encoding="utf-8", ) as f: diff --git a/utils/generate_enums.py b/utils/generate_enums.py index 3b88d9be..b8f039a3 100644 --- a/utils/generate_enums.py +++ b/utils/generate_enums.py @@ -12,13 +12,15 @@ import re import subprocess from pathlib import Path -from typing import cast +from typing import TYPE_CHECKING, cast from pyoverkiz.auth.credentials import UsernamePasswordCredentials from pyoverkiz.client import OverkizClient from pyoverkiz.enums import Server from pyoverkiz.exceptions import OverkizError -from pyoverkiz.models import UIProfileDefinition, ValuePrototype + +if TYPE_CHECKING: + from pyoverkiz.models import UIProfileDefinition, ValuePrototype # Hardcoded protocols that may not be available on all servers # Each tuple contains: name, prefix, id, label @@ -129,8 +131,8 @@ async def generate_ui_enums(server: Server) -> None: ) as client: await client.login() - ui_classes = cast(list[str], await client.get_reference_ui_classes()) - ui_widgets = cast(list[str], await client.get_reference_ui_widgets()) + ui_classes = cast("list[str]", await client.get_reference_ui_classes()) + ui_widgets = cast("list[str]", await client.get_reference_ui_widgets()) # Convert camelCase to SCREAMING_SNAKE_CASE for enum names def to_enum_name(value: str) -> str: @@ -210,7 +212,7 @@ def to_enum_name(value: str) -> str: lines.append("") # End with newline # Fetch and add UI classifiers - ui_classifiers = cast(list[str], await client.get_reference_ui_classifiers()) + ui_classifiers = cast("list[str]", await client.get_reference_ui_classifiers()) lines.append("") lines.append("@unique")