Migrate to cattrs for centralized model structuring

Replace per-model converter functions (_flexible_init, _to_list,
_to_optional, _to_optional_enum) with a single cattrs Converter that
handles all dict-to-model structuring. Models become pure data classes
with only type annotations and defaults.

- Add cattrs dependency
- Add pyoverkiz/converter.py with hooks for primitive unions, optional
  attrs classes, enums, and container types (States, CommandDefinitions)
- Strip all converter infrastructure from models.py
- Update client.py to use converter.structure() instead of Model(**data)
- Update tests to use converter for API-like dict structuring

Author: iMicknl

pyoverkiz/client.py

@@ -22,6 +22,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, CommandMode, Server
 from pyoverkiz.exceptions import (
     ExecutionQueueFullError,
@@ -305,7 +306,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
@@ -343,7 +344,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
@@ -362,7 +363,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
@@ -375,7 +376,7 @@ async def get_gateways(self, refresh: bool = False) -> list[Gateway]:
     async def get_execution_history(self) -> list[HistoryExecution]:
         """List execution history."""
         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:
@@ -392,7 +393,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:
@@ -436,7 +437,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.
@@ -451,13 +452,13 @@ async def unregister_event_listener(self) -> None:
     async def get_current_execution(self, exec_id: str) -> Execution:
         """Get an action group execution currently running."""
         response = await self._get(f"exec/current/{exec_id}")
-        return Execution(**decamelize(response))
+        return converter.structure(decamelize(response), Execution)
 
     @retry_on_auth_error
     async def get_current_executions(self) -> list[Execution]:
         """Get all action groups executions currently running."""
         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:
@@ -555,7 +556,7 @@ async def cancel_command(self, exec_id: str) -> None:
     async def get_action_groups(self) -> list[ActionGroup]:
         """List the action groups (scenarios)."""
         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:
@@ -570,7 +571,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_scenario(self, oid: str) -> str:
@@ -592,7 +593,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:
@@ -603,7 +604,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
 
@@ -621,7 +622,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
 
@@ -653,7 +654,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:
@@ -683,7 +684,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]:

pyoverkiz/converter.py

@@ -0,0 +1,118 @@
+"""Centralized cattrs converter for structuring Overkiz API responses into models."""
+
+from __future__ import annotations
+
+import types
+from enum import Enum
+from typing import Any, Union, get_args, get_origin
+
+import attr
+import cattrs
+from cattrs.dispatch import StructureHook
+
+from pyoverkiz.models import (
+    CommandDefinition,
+    CommandDefinitions,
+    State,
+    States,
+)
+
+
+def _is_union(t: Any) -> bool:
+    """Check if a type is a Union or PEP 604 union (X | Y)."""
+    return get_origin(t) is Union or isinstance(t, types.UnionType)
+
+
+def _is_primitive_union(t: Any) -> bool:
+    """Check if a union type only contains primitive/JSON-native types and enums.
+
+    Returns False if any union member is an attrs class (needs structuring).
+    """
+    if not _is_union(t):
+        return False
+    for arg in get_args(t):
+        if arg is type(None):
+            continue
+        if isinstance(arg, type) and attr.has(arg):
+            return False
+    return True
+
+
+def _make_converter() -> cattrs.Converter:
+    c = cattrs.Converter(forbid_extra_keys=False)
+
+    # ------------------------------------------------------------------
+    # Primitive union types (str | Enum, StateType, etc.): passthrough
+    # These only contain JSON-native types and enums, no attrs classes.
+    # ------------------------------------------------------------------
+    c.register_structure_hook_func(_is_primitive_union, lambda v, _: v)
+
+    # ------------------------------------------------------------------
+    # Optional[AttrsClass] unions: structure the non-None member
+    # ------------------------------------------------------------------
+    def _is_optional_attrs(t: Any) -> bool:
+        if not _is_union(t):
+            return False
+        args = get_args(t)
+        non_none = [a for a in args if a is not type(None)]
+        return (
+            len(non_none) == 1
+            and isinstance(non_none[0], type)
+            and attr.has(non_none[0])
+        )
+
+    def _structure_optional_attrs(v: Any, t: Any) -> Any:
+        if v is None:
+            return None
+        args = get_args(t)
+        cls = next(a for a in args if a is not type(None))
+        return c.structure(v, cls)
+
+    c.register_structure_hook_func(_is_optional_attrs, _structure_optional_attrs)
+
+    # ------------------------------------------------------------------
+    # Enums: use the enum constructor which handles UnknownEnumMixin
+    # ------------------------------------------------------------------
+    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),
+    )
+
+    # ------------------------------------------------------------------
+    # attrs classes: silently discard unknown keys from API responses
+    # ------------------------------------------------------------------
+    def _make_attrs_hook(cls: type) -> StructureHook:
+        inner: StructureHook = cattrs.gen.make_dict_structure_fn(cls, c)
+
+        def hook(val: Any, _: type) -> Any:
+            if isinstance(val, dict):
+                fields = {a.name for a in attr.fields(cls)}
+                val = {k: v for k, v in val.items() if k in fields}
+            return inner(val, cls)
+
+        return hook
+
+    c.register_structure_hook_factory(attr.has, _make_attrs_hook)
+
+    # ------------------------------------------------------------------
+    # Container types with custom __init__
+    # ------------------------------------------------------------------
+    def _structure_states(val: Any, _: type) -> States:
+        if isinstance(val, States):
+            return val
+        if val is None:
+            return States()
+        return States([c.structure(s, State) for s in val])
+
+    def _structure_command_definitions(val: Any, _: type) -> CommandDefinitions:
+        if isinstance(val, CommandDefinitions):
+            return val
+        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()