feat(teslemetry): make authorized-clients parser public, add raw option

Rename _parse_authorized_clients to parse_authorized_clients and export it
alongside AuthorizedClients/AuthorizedClient and the related enums from
tesla_fleet_api.teslemetry, so a local aiopowerwall reader can reuse the
same parser instead of duplicating it. find_authorized_clients() gains
raw=False to return the same unparsed dict as list_authorized_clients()
for callers that want the raw shape. The private name stays as an alias
for one release.

Author: Bre77

pyproject.toml

@@ -4,7 +4,7 @@ requires = ["setuptools>=77.0"]
 
 [project]
 name = "tesla_fleet_api"
-version = "1.13.0"
+version = "1.14.0"
 license = "Apache-2.0"
 description = "Tesla Fleet API library for Python"
 readme = "README.md"

tesla_fleet_api/__init__.py

@@ -1,7 +1,7 @@
 """Tesla Fleet API"""
 
 __author__ = "hello@teslemetry.com"
-__version__ = "1.13.0"
+__version__ = "1.14.0"
 
 from tesla_fleet_api.const import Region, is_valid_region
 from tesla_fleet_api.funnel import (

tesla_fleet_api/teslemetry/__init__.py

@@ -1,6 +1,18 @@
+from tesla_fleet_api.const import (
+    AuthorizationRole,
+    AuthorizedClientKeyType,
+    AuthorizedClientState,
+    AuthorizedClientType,
+    AuthorizedVerificationType,
+)
 from tesla_fleet_api.tesla.charging import Charging
 from tesla_fleet_api.tesla.energysite import EnergySite, EnergySites
 from tesla_fleet_api.tesla.user import User
+from tesla_fleet_api.teslemetry.energysite import (
+    AuthorizedClient,
+    AuthorizedClients,
+    parse_authorized_clients,
+)
 from tesla_fleet_api.teslemetry.teslemetry import (
     Teslemetry,
     TeslemetryClientRegistration,
@@ -13,6 +25,14 @@
     "Teslemetry",
     "TeslemetryClientRegistration",
     "register_client",
+    "AuthorizationRole",
+    "AuthorizedClient",
+    "AuthorizedClientKeyType",
+    "AuthorizedClients",
+    "AuthorizedClientState",
+    "AuthorizedClientType",
+    "AuthorizedVerificationType",
+    "parse_authorized_clients",
     "Charging",
     "EnergySites",
     "EnergySite",

tesla_fleet_api/teslemetry/energysite.py

@@ -7,7 +7,7 @@
 import struct
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
-from typing import Any, cast
+from typing import Any, Literal, overload, cast
 
 from tesla_fleet_api.const import (
     AuthorizationRole,
@@ -191,9 +191,10 @@ def _authorized_clients_list(payload: Any) -> list[Any]:
     return cast("list[Any]", value)
 
 
-def _parse_authorized_clients(payload: Any) -> AuthorizedClients:
-    """Parse a raw ``list_authorized_clients()`` response into typed clients.
+def parse_authorized_clients(payload: Any) -> AuthorizedClients:
+    """The one parser for a Teslemetry/aiopowerwall authorized-clients cloud envelope.
 
+    Parses a raw ``list_authorized_clients()`` response into typed clients.
     Raises :class:`~tesla_fleet_api.exceptions.InvalidResponse` if
     ``payload`` is null or doesn't match the confirmed envelope shape - see
     :func:`_authorized_clients_list`.
@@ -206,6 +207,10 @@ def _parse_authorized_clients(payload: Any) -> AuthorizedClients:
     return AuthorizedClients(clients=clients, raw=payload)
 
 
+# Deprecated alias, kept for one release.
+_parse_authorized_clients = parse_authorized_clients
+
+
 _GATEWAY_INTERFACES = ("eth", "wifi")
 
 _DOTTED_QUAD_OCTET = r"(25[0-5]|2[0-4][0-9]|1[0-9]{2}|[1-9]?[0-9])"
@@ -392,7 +397,15 @@ async def list_authorized_clients(self) -> dict[str, Any]:
             f"api/1/energy_sites/{self.energy_site_id}/command/authorized_clients",
         )
 
-    async def find_authorized_clients(self) -> AuthorizedClients:
+    @overload
+    async def find_authorized_clients(
+        self, raw: Literal[False] = False
+    ) -> AuthorizedClients: ...
+    @overload
+    async def find_authorized_clients(self, raw: Literal[True]) -> dict[str, Any]: ...
+    async def find_authorized_clients(
+        self, raw: bool = False
+    ) -> AuthorizedClients | dict[str, Any]:
         """List authorized clients on the energy gateway, parsed into a typed result.
 
         Prefer this over :meth:`list_authorized_clients` for consumers that
@@ -403,8 +416,16 @@ async def find_authorized_clients(self) -> AuthorizedClients:
         response body or an unrecognized response shape rather than
         treating either as "no clients". See :class:`AuthorizedClients` for
         the exact parsing semantics.
+
+        ``raw=True`` returns the unparsed response exactly as
+        :meth:`list_authorized_clients` does, skipping the typed parse - for
+        callers that want the same call shape aligned with the local
+        gateway path while still opting out of typing.
         """
-        return _parse_authorized_clients(await self.list_authorized_clients())
+        response = await self.list_authorized_clients()
+        if raw:
+            return response
+        return parse_authorized_clients(response)
 
     async def remove_authorized_client(self, public_key: bytes | str) -> dict[str, Any]:
         """Remove an authorized client from the energy gateway via the

tests/test_teslemetry_authorized_clients.py

@@ -28,6 +28,12 @@
     AuthorizedVerificationType,
 )
 from tesla_fleet_api.exceptions import InvalidResponse
+from tesla_fleet_api.teslemetry import (
+    AuthorizedClient,
+    AuthorizedClients,
+    parse_authorized_clients,
+)
+from tesla_fleet_api.teslemetry.energysite import _parse_authorized_clients
 from tesla_fleet_api.teslemetry.teslemetry import Teslemetry
 
 _UNSET = object()
@@ -66,7 +72,59 @@ def _make_site(json_body: object):
     return api.energySites.create(12345)
 
 
+class PublicParserImportTests(IsolatedAsyncioTestCase):
+    async def test_parser_is_importable_from_public_teslemetry_path(self) -> None:
+        payload = {
+            "response": {
+                "authorized_clients": [
+                    {"public_key": PUBLIC_KEY_B64, "state": 3},
+                ]
+            }
+        }
+
+        result = parse_authorized_clients(payload)
+
+        self.assertIsInstance(result, AuthorizedClients)
+        self.assertIsInstance(result.clients[0], AuthorizedClient)
+
+    async def test_private_alias_still_works_for_one_release(self) -> None:
+        payload = {"response": {"authorized_clients": []}}
+
+        self.assertEqual(
+            _parse_authorized_clients(payload), parse_authorized_clients(payload)
+        )
+
+
 class GetAuthorizedClientsTests(IsolatedAsyncioTestCase):
+    async def test_raw_true_returns_unparsed_response(self) -> None:
+        payload = {
+            "response": {
+                "authorized_clients": [
+                    {"public_key": PUBLIC_KEY_B64, "state": 3},
+                ]
+            }
+        }
+        site = _make_site(payload)
+
+        result = await site.find_authorized_clients(raw=True)
+
+        self.assertEqual(result, payload)
+
+    async def test_default_raw_false_behaviour_unchanged(self) -> None:
+        payload = {
+            "response": {
+                "authorized_clients": [
+                    {"public_key": PUBLIC_KEY_B64, "state": 3},
+                ]
+            }
+        }
+        site = _make_site(payload)
+
+        result = await site.find_authorized_clients()
+
+        self.assertEqual(len(result.clients), 1)
+        self.assertEqual(result.clients[0].public_key, PUBLIC_KEY_B64)
+
     async def test_normal_payload_round_trips(self) -> None:
         site = _make_site(
             {