Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions pyoverkiz/_case.py
Original file line number Diff line number Diff line change
Expand Up @@ -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])")

Expand Down
3 changes: 2 additions & 1 deletion pyoverkiz/action_queue.py
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand Down
6 changes: 4 additions & 2 deletions pyoverkiz/auth/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
12 changes: 8 additions & 4 deletions pyoverkiz/auth/factory.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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(
Expand Down
26 changes: 14 additions & 12 deletions pyoverkiz/auth/strategies.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -48,7 +51,6 @@
SomfyBadCredentialsError,
SomfyServiceError,
)
from pyoverkiz.models import ServerConfig

MIN_JWT_SEGMENTS = 2

Expand Down Expand Up @@ -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

Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
24 changes: 14 additions & 10 deletions pyoverkiz/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
Expand All @@ -17,7 +16,6 @@
ClientSession,
ServerDisconnectedError,
)
from backoff.types import Details

from pyoverkiz._case import decamelize
from pyoverkiz.action_queue import ActionQueue, ActionQueueSettings
Expand Down Expand Up @@ -53,14 +51,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:
Expand Down Expand Up @@ -417,7 +421,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
Expand Down Expand Up @@ -463,7 +467,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
Expand All @@ -483,7 +487,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,
Expand Down Expand Up @@ -569,13 +573,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]:
Expand Down
2 changes: 1 addition & 1 deletion pyoverkiz/enums/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -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]
27 changes: 15 additions & 12 deletions pyoverkiz/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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

# <protocol>://<gatewayId>/<deviceAddress>[#<subsystemId>]
Expand Down Expand Up @@ -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
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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")


Expand Down Expand Up @@ -891,7 +894,7 @@ def __init__(
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.id = cast("str", oid or id)
self.creation_time = creation_time
self.last_update_time = last_update_time
self.label = (
Expand All @@ -904,7 +907,7 @@ def __init__(
self.notification_text = notification_text
self.notification_title = notification_title
self.actions = [Action(**action) for action in actions]
self.oid = cast(str, oid or id)
self.oid = cast("str", oid or id)


@define(init=False, kw_only=True)
Expand Down
5 changes: 3 additions & 2 deletions pyoverkiz/obfuscate.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
6 changes: 4 additions & 2 deletions pyoverkiz/response_handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -37,6 +36,9 @@
UnknownUserError,
)

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.
Expand Down
6 changes: 4 additions & 2 deletions pyoverkiz/types.py
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand Down
3 changes: 3 additions & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -124,6 +124,8 @@ select = [
"A",
# pylint
"PL",
# flake8-type-checking
"TC",
]
ignore = [
"E501", # Line too long
Expand All @@ -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]
Expand Down
Loading
Loading