Migrate to cattrs for centralized model structuring

pyoverkiz/auth/base.py

@@ -4,16 +4,16 @@
 
 import datetime
 from collections.abc import Mapping
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from typing import Any, Protocol
 
 
 @dataclass(slots=True)
 class AuthContext:
     """Authentication context holding tokens and expiration."""
 
-    access_token: str | None = None
-    refresh_token: str | None = None
+    access_token: str | None = field(default=None, repr=False)
+    refresh_token: str | None = field(default=None, repr=False)
     expires_at: datetime.datetime | None = None
 
     def is_expired(self, *, skew_seconds: int = 5) -> bool:

pyoverkiz/auth/credentials.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 
 
 class Credentials:
@@ -14,14 +14,14 @@ class UsernamePasswordCredentials(Credentials):
     """Credentials using username and password."""
 
     username: str
-    password: str
+    password: str = field(repr=False)
 
 
 @dataclass(slots=True)
 class TokenCredentials(Credentials):
     """Credentials using an (API) token."""
 
-    token: str
+    token: str = field(repr=False)
 
 
 @dataclass(slots=True)
@@ -33,5 +33,5 @@ class LocalTokenCredentials(TokenCredentials):
 class RexelOAuthCodeCredentials(Credentials):
     """Credentials using Rexel OAuth2 authorization code."""
 
-    code: str
+    code: str = field(repr=False)
     redirect_uri: str

pyoverkiz/client.py

@@ -23,6 +23,7 @@
 from pyoverkiz.action_queue import ActionQueue, ActionQueueSettings
 from pyoverkiz.auth import AuthStrategy, Credentials, build_auth_strategy
 from pyoverkiz.const import SUPPORTED_SERVERS
+from pyoverkiz.converter import converter
 from pyoverkiz.enums import APIType, ExecutionMode, Server
 from pyoverkiz.exceptions import (
     ExecutionQueueFullError,
@@ -306,7 +307,7 @@ async def get_setup(self, refresh: bool = False) -> Setup:
 
         response = await self._get("setup")
 
-        setup = Setup(**decamelize(response))
+        setup = converter.structure(decamelize(response), Setup)
 
         # Cache response
         self.setup = setup
@@ -344,7 +345,7 @@ async def get_devices(self, refresh: bool = False) -> list[Device]:
             return self.devices
 
         response = await self._get("setup/devices")
-        devices = [Device(**d) for d in decamelize(response)]
+        devices = converter.structure(decamelize(response), list[Device])
 
         # Cache response
         self.devices = devices
@@ -363,7 +364,7 @@ async def get_gateways(self, refresh: bool = False) -> list[Gateway]:
             return self.gateways
 
         response = await self._get("setup/gateways")
-        gateways = [Gateway(**g) for g in decamelize(response)]
+        gateways = converter.structure(decamelize(response), list[Gateway])
 
         # Cache response
         self.gateways = gateways
@@ -376,7 +377,7 @@ async def get_gateways(self, refresh: bool = False) -> list[Gateway]:
     async def get_execution_history(self) -> list[HistoryExecution]:
         """List past executions and their outcomes."""
         response = await self._get("history/executions")
-        return [HistoryExecution(**h) for h in decamelize(response)]
+        return converter.structure(decamelize(response), list[HistoryExecution])
 
     @retry_on_auth_error
     async def get_device_definition(self, deviceurl: str) -> JSON | None:
@@ -393,7 +394,7 @@ async def get_state(self, deviceurl: str) -> list[State]:
         response = await self._get(
             f"setup/devices/{urllib.parse.quote_plus(deviceurl)}/states"
         )
-        return [State(**s) for s in decamelize(response)]
+        return converter.structure(decamelize(response), list[State])
 
     @retry_on_auth_error
     async def refresh_states(self) -> None:
@@ -437,7 +438,7 @@ async def fetch_events(self) -> list[Event]:
         """
         await self._refresh_token_if_expired()
         response = await self._post(f"events/{self.event_listener_id}/fetch")
-        return [Event(**e) for e in decamelize(response)]
+        return converter.structure(decamelize(response), list[Event])
 
     async def unregister_event_listener(self) -> None:
         """Unregister an event listener.
@@ -455,17 +456,16 @@ async def get_current_execution(self, exec_id: str) -> Execution | None:
         Returns None if the execution does not exist.
         """
         response = await self._get(f"exec/current/{exec_id}")
-
         if not response or not isinstance(response, dict):
             return None
 
-        return Execution(**decamelize(response))
+        return converter.structure(decamelize(response), Execution)
 
     @retry_on_auth_error
     async def get_current_executions(self) -> list[Execution]:
         """Get all currently running executions."""
         response = await self._get("exec/current")
-        return [Execution(**e) for e in decamelize(response)]
+        return converter.structure(decamelize(response), list[Execution])
 
     @retry_on_auth_error
     async def get_api_version(self) -> str:
@@ -557,7 +557,7 @@ async def cancel_execution(self, exec_id: str) -> None:
     async def get_action_groups(self) -> list[ActionGroup]:
         """List action groups persisted on the server."""
         response = await self._get("actionGroups")
-        return [ActionGroup(**action_group) for action_group in decamelize(response)]
+        return converter.structure(decamelize(response), list[ActionGroup])
 
     @retry_on_auth_error
     async def get_places(self) -> Place:
@@ -572,7 +572,7 @@ async def get_places(self) -> Place:
         - `sub_places`: List of nested places within this location
         """
         response = await self._get("setup/places")
-        return Place(**decamelize(response))
+        return converter.structure(decamelize(response), Place)
 
     @retry_on_auth_error
     async def execute_persisted_action_group(self, oid: str) -> str:
@@ -594,7 +594,7 @@ async def get_setup_options(self) -> list[Option]:
         Access scope : Full enduser API access (enduser/*).
         """
         response = await self._get("setup/options")
-        return [Option(**o) for o in decamelize(response)]
+        return converter.structure(decamelize(response), list[Option])
 
     @retry_on_auth_error
     async def get_setup_option(self, option: str) -> Option | None:
@@ -605,7 +605,7 @@ async def get_setup_option(self, option: str) -> Option | None:
         response = await self._get(f"setup/options/{option}")
 
         if response:
-            return Option(**decamelize(response))
+            return converter.structure(decamelize(response), Option)
 
         return None
 
@@ -623,7 +623,7 @@ async def get_setup_option_parameter(
         response = await self._get(f"setup/options/{option}/{parameter}")
 
         if response:
-            return OptionParameter(**decamelize(response))
+            return converter.structure(decamelize(response), OptionParameter)
 
         return None
 
@@ -655,7 +655,7 @@ async def get_reference_protocol_types(self) -> list[ProtocolType]:
         - label: Human-readable protocol label
         """
         response = await self._get("reference/protocolTypes")
-        return [ProtocolType(**protocol) for protocol in response]
+        return converter.structure(response, list[ProtocolType])
 
     @retry_on_auth_error
     async def get_reference_timezones(self) -> JSON:
@@ -685,7 +685,7 @@ async def get_reference_ui_profile(self, profile_name: str) -> UIProfileDefiniti
         response = await self._get(
             f"reference/ui/profile/{urllib.parse.quote_plus(profile_name)}"
         )
-        return UIProfileDefinition(**decamelize(response))
+        return converter.structure(decamelize(response), UIProfileDefinition)
 
     @retry_on_auth_error
     async def get_reference_ui_profile_names(self) -> list[str]:
@@ -701,7 +701,7 @@ async def get_reference_ui_widgets(self) -> list[str]:
     async def get_devices_not_up_to_date(self) -> list[Device]:
         """Get all devices whose firmware is not up to date."""
         response = await self._get("setup/devices/notUpToDate")
-        return [Device(**d) for d in decamelize(response)]
+        return converter.structure(decamelize(response), list[Device])
 
     @retry_on_auth_error
     async def get_device_firmware_status(self, deviceurl: str) -> FirmwareStatus | None:
@@ -715,7 +715,7 @@ async def get_device_firmware_status(self, deviceurl: str) -> FirmwareStatus | N
             )
         except UnsupportedOperationError:
             return None
-        return FirmwareStatus(**decamelize(response))
+        return converter.structure(decamelize(response), FirmwareStatus)
 
     @retry_on_auth_error
     async def get_device_firmware_update_capability(self, deviceurl: str) -> bool:

pyoverkiz/converter.py

@@ -0,0 +1,64 @@
+"""Centralized cattrs converter for structuring Overkiz API responses."""
+
+from __future__ import annotations
+
+import types
+from enum import Enum
+from typing import Any, Union, get_args, get_origin
+
+import attr
+import cattrs
+
+from pyoverkiz.models import (
+    CommandDefinition,
+    CommandDefinitions,
+    State,
+    States,
+)
+
+
+def _is_primitive_union(t: Any) -> bool:
+    """True for unions of JSON-native types (e.g. StateType).
+
+    Excludes unions containing attrs classes (e.g. Definition | None) since those
+    need actual structuring by cattrs.
+    """
+    origin = get_origin(t)
+    if origin is not Union and not isinstance(t, types.UnionType):
+        return False
+    non_none = [arg for arg in get_args(t) if arg is not type(None)]
+    if any(isinstance(arg, type) and attr.has(arg) for arg in non_none):
+        return False
+    # Exclude pure Optional[Enum] unions — those need the Enum structure hook.
+    return not all(isinstance(arg, type) and issubclass(arg, Enum) for arg in non_none)
+
+
+def _make_converter() -> cattrs.Converter:
+    c = cattrs.Converter()
+
+    # JSON-native unions like StateType (str | int | float | … | None) are already the
+    # correct Python type after JSON parsing — tell cattrs to pass them through as-is.
+    c.register_structure_hook_func(_is_primitive_union, lambda v, _: v)
+
+    # Enums: call the constructor so UnknownEnumMixin._missing_ can handle unknown values
+    c.register_structure_hook_func(
+        lambda t: isinstance(t, type) and issubclass(t, Enum),
+        lambda v, t: v if isinstance(v, t) else t(v),
+    )
+
+    # Custom container types that take a list in __init__
+    def _structure_states(val: Any, _: type) -> States:
+        if val is None:
+            return States()
+        return States([c.structure(s, State) for s in val])
+
+    def _structure_command_definitions(val: Any, _: type) -> CommandDefinitions:
+        return CommandDefinitions([c.structure(cd, CommandDefinition) for cd in val])
+
+    c.register_structure_hook(States, _structure_states)
+    c.register_structure_hook(CommandDefinitions, _structure_command_definitions)
+
+    return c
+
+
+converter = _make_converter()

pyoverkiz/enums/__init__.py

@@ -7,7 +7,12 @@
     ExecutionSubType,
     ExecutionType,
 )
-from pyoverkiz.enums.gateway import GatewaySubType, GatewayType, UpdateBoxStatus
+from pyoverkiz.enums.gateway import (
+    GatewaySubType,
+    GatewayType,
+    UpdateBoxStatus,
+    UpdateCriticityLevel,
+)
 from pyoverkiz.enums.general import DataType, EventName, FailureType, ProductType
 from pyoverkiz.enums.measured_value_type import MeasuredValueType
 from pyoverkiz.enums.protocol import Protocol
@@ -40,4 +45,5 @@
     "UIProfile",
     "UIWidget",
     "UpdateBoxStatus",
+    "UpdateCriticityLevel",
 ]

pyoverkiz/enums/gateway.py

@@ -107,6 +107,15 @@ def beautify_name(self) -> str:
         )
 
 
+@unique
+class UpdateCriticityLevel(UnknownEnumMixin, StrEnum):
+    """Criticity level of an available gateway update."""
+
+    BUG_FIX = "BUG_FIX"
+    DEVICES_CONTROL_ONLY = "DEVICES_CONTROL_ONLY"
+    UNKNOWN = "UNKNOWN"
+
+
 @unique
 class UpdateBoxStatus(StrEnum):
     """Status of the gateway update box indicating its updateability."""