From 0d67bae8969e2500b0f3069bdf8f7e0f4aaed782 Mon Sep 17 00:00:00 2001 From: Brett Date: Sat, 1 Jun 2024 13:16:39 +1000 Subject: [PATCH] So many fixes --- README.md | 4 -- tesla_fleet_api/energy.py | 8 +++- tesla_fleet_api/energyspecific.py | 11 ++++- tesla_fleet_api/exceptions.py | 8 ++-- tesla_fleet_api/teslafleetapi.py | 35 +++++++--------- tesla_fleet_api/teslafleetoauth.py | 8 ++-- tesla_fleet_api/teslemetry.py | 9 ++--- tesla_fleet_api/tessie.py | 2 - tesla_fleet_api/vehicle.py | 64 ++++++++---------------------- tesla_fleet_api/vehiclespecific.py | 41 +++++-------------- 10 files changed, 69 insertions(+), 121 deletions(-) diff --git a/README.md b/README.md index 34d4f2e..6fd751a 100644 --- a/README.md +++ b/README.md @@ -23,7 +23,6 @@ async def main(): access_token="", session=session, region="na", - raise_for_status=True, ) try: @@ -54,7 +53,6 @@ async def main(): refresh_token=auth["refresh_token"], expires=auth["expires"], region="na", - raise_for_status=True, ) try: data = await api.vehicle.list() @@ -91,7 +89,6 @@ async def main(): api = Teslemetry( access_token="", session=session, - raise_for_status=True, ) try: @@ -119,7 +116,6 @@ async def main(): api = Tessie( access_token="", session=session, - raise_for_status=True, ) try: diff --git a/tesla_fleet_api/energy.py b/tesla_fleet_api/energy.py index 85df226..86ff643 100644 --- a/tesla_fleet_api/energy.py +++ b/tesla_fleet_api/energy.py @@ -1,11 +1,17 @@ -from typing import Any +from __future__ import annotations +from typing import Any, TYPE_CHECKING from .const import Method, EnergyOperationMode, EnergyExportMode from .energyspecific import EnergySpecific +if TYPE_CHECKING: + from .teslafleetapi import TeslaFleetApi + class Energy: """Class describing the Tesla Fleet API partner endpoints""" + parent: TeslaFleetApi + def __init__(self, parent): self._request = parent._request diff --git a/tesla_fleet_api/energyspecific.py b/tesla_fleet_api/energyspecific.py index e78a5f7..68c2ee8 100644 --- a/tesla_fleet_api/energyspecific.py +++ b/tesla_fleet_api/energyspecific.py @@ -1,13 +1,20 @@ -from typing import Any +from __future__ import annotations +from typing import Any, TYPE_CHECKING from .const import EnergyExportMode, EnergyOperationMode +if TYPE_CHECKING: + from .energy import Energy + class EnergySpecific: """Class describing the Tesla Fleet API partner endpoints""" + _parent: Energy + energy_site_id: int + def __init__( self, - parent, + parent: Energy, energy_site_id: int, ): self._parent = parent diff --git a/tesla_fleet_api/exceptions.py b/tesla_fleet_api/exceptions.py index 4a352c2..020124a 100644 --- a/tesla_fleet_api/exceptions.py +++ b/tesla_fleet_api/exceptions.py @@ -7,9 +7,10 @@ class TeslaFleetError(BaseException): message: str = "An unknown error has occurred." status: int | None = None - data: dict | None = None + data: dict | str | None = None + key: str | None = None - def __init__(self, data: dict | None = None, status: int | None = None): + def __init__(self, data: dict | str | None = None, status: int | None = None): LOGGER.debug(self.message) self.data = data self.status = status or self.status @@ -20,6 +21,7 @@ class ResponseError(TeslaFleetError): """The response from the server was not JSON.""" message = "The response from the server was not JSON." + data: str | None = None class InvalidCommand(TeslaFleetError): @@ -375,5 +377,5 @@ async def raise_for_status(resp: aiohttp.ClientResponse) -> None: elif resp.status == 540: raise DeviceUnexpectedResponse(data) elif data is None: - raise ResponseError(status=resp.status) + raise ResponseError(status=resp.status, data=await resp.text()) resp.raise_for_status() diff --git a/tesla_fleet_api/teslafleetapi.py b/tesla_fleet_api/teslafleetapi.py index 44268e9..2bc554f 100644 --- a/tesla_fleet_api/teslafleetapi.py +++ b/tesla_fleet_api/teslafleetapi.py @@ -1,8 +1,10 @@ -import aiohttp +"""Tesla Fleet API for Python.""" from json import dumps -from .exceptions import raise_for_status, InvalidRegion, LibraryError, InvalidToken from typing import Any +import aiohttp + +from .exceptions import raise_for_status, InvalidRegion, LibraryError, ResponseError from .const import SERVERS, Method, LOGGER, VERSION from .charging import Charging from .energy import Energy @@ -15,10 +17,11 @@ class TeslaFleetApi: """Class describing the Tesla Fleet API.""" + access_token: str | None = None + region: str | None = None server: str | None = None session: aiohttp.ClientSession headers: dict[str, str] - raise_for_status: bool def __init__( self, @@ -26,7 +29,6 @@ def __init__( access_token: str | None = None, region: str | None = None, server: str | None = None, - raise_for_status: bool = True, charging_scope: bool = True, energy_scope: bool = True, partner_scope: bool = True, @@ -37,7 +39,6 @@ def __init__( self.session = session self.access_token = access_token - self.raise_for_status = raise_for_status if server is not None: self.server = server @@ -82,7 +83,7 @@ async def _request( path: str, params: dict[str, Any] | None = None, json: dict[str, Any] | None = None, - ) -> dict[str, Any] | str: + ) -> dict[str, Any]: """Send a request to the Tesla Fleet API.""" if not self.server: @@ -113,22 +114,14 @@ async def _request( params=params, ) as resp: LOGGER.debug("Response Status: %s", resp.status) - if self.raise_for_status and not resp.ok: + if not resp.ok: await raise_for_status(resp) - elif resp.status == 401 and resp.content_type != "application/json": - # Manufacture a response since Tesla doesn't provide a body for token expiration. - return { - "response": None, - "error": InvalidToken.key, - "error_message": "The OAuth token has expired.", - } - if resp.content_type == "application/json": - data = await resp.json() - LOGGER.debug("Response JSON: %s", data) - return data - - data = await resp.text() - LOGGER.debug("Response Text: %s", data) + if not resp.content_type.lower().startswith("application/json"): + LOGGER.debug("Response type is: %s", resp.content_type) + raise ResponseError(status=resp.status, data=await resp.text()) + + data = await resp.json() + LOGGER.debug("Response JSON: %s", data) return data async def status(self) -> str: diff --git a/tesla_fleet_api/teslafleetoauth.py b/tesla_fleet_api/teslafleetoauth.py index 631eee3..2604902 100644 --- a/tesla_fleet_api/teslafleetoauth.py +++ b/tesla_fleet_api/teslafleetoauth.py @@ -21,7 +21,6 @@ def __init__( expires: int = 0, region: str | None = None, server: str | None = None, - raise_for_status: bool = True, ): self.client_id = client_id self.access_token = access_token @@ -33,7 +32,6 @@ def __init__( access_token="", region=region, server=server, - raise_for_status=raise_for_status, ) def get_login_url( @@ -96,8 +94,8 @@ async def _request( method: Method, path: str, params: dict[str, Any] | None = None, - data: dict[str, Any] | None = None, - ) -> str | dict[str, Any]: + json: dict[str, Any] | None = None, + ) -> dict[str, Any]: """Send a request to the Tesla Fleet API.""" await self.check_access_token() - return await super()._request(method, path, params, data) + return await super()._request(method, path, params, json) diff --git a/tesla_fleet_api/teslemetry.py b/tesla_fleet_api/teslemetry.py index b6630ff..926ae83 100644 --- a/tesla_fleet_api/teslemetry.py +++ b/tesla_fleet_api/teslemetry.py @@ -13,14 +13,12 @@ def __init__( self, session: aiohttp.ClientSession, access_token: str, - raise_for_status: bool = True, ): """Initialize the Teslemetry API.""" super().__init__( session, access_token, server="https://api.teslemetry.com", - raise_for_status=raise_for_status, partner_scope=False, user_scope=False, ) @@ -52,10 +50,11 @@ async def metadata(self, update_region=True) -> dict[str, Any]: LOGGER.debug("Using server %s", self.server) return resp - # TODO: type this properly, it probably should return something - async def find_server(self) -> None: + async def find_server(self) -> str: """Find the server URL for the Tesla Fleet API.""" await self.metadata(True) + assert self.region + return self.region async def _request( self, @@ -63,7 +62,7 @@ async def _request( path: str, params: dict[str, Any] | None = None, json: dict[str, Any] | None = None, - ) -> str | dict[str, Any]: + ) -> dict[str, Any]: """Send a request to the Teslemetry API.""" async with rate_limit: return await super()._request(method, path, params, json) diff --git a/tesla_fleet_api/tessie.py b/tesla_fleet_api/tessie.py index 999f907..5816bc6 100644 --- a/tesla_fleet_api/tessie.py +++ b/tesla_fleet_api/tessie.py @@ -9,14 +9,12 @@ def __init__( self, session: aiohttp.ClientSession, access_token: str, - raise_for_status: bool = True, ): """Initialize the Tessie API.""" super().__init__( session, access_token, server="https://api.tessie.com", - raise_for_status=raise_for_status, partner_scope=False, user_scope=False, ) diff --git a/tesla_fleet_api/vehicle.py b/tesla_fleet_api/vehicle.py index 6534e5b..5e3f9d5 100644 --- a/tesla_fleet_api/vehicle.py +++ b/tesla_fleet_api/vehicle.py @@ -1,4 +1,5 @@ -from typing import Any, List +from __future__ import annotations +from typing import Any, List, TYPE_CHECKING from .const import ( Method, Trunk, @@ -13,17 +14,22 @@ ) from .vehiclespecific import VehicleSpecific +if TYPE_CHECKING: + from .teslafleetapi import TeslaFleetApi + class Vehicle: """Class describing the Tesla Fleet API vehicle endpoints and commands.""" - def __init__(self, parent): + _parent: TeslaFleetApi + + def __init__(self, parent: TeslaFleetApi): self._parent = parent self._request = parent._request - def specific(self, vehicle_tag: str | int) -> VehicleSpecific: + def specific(self, vin: str) -> VehicleSpecific: """Creates a class for each vehicle.""" - return VehicleSpecific(self, vehicle_tag) + return VehicleSpecific(self, vin) def pre2021(self, vin: str) -> bool: """Checks if a vehicle is a pre-2021 model S or X.""" @@ -716,26 +722,6 @@ async def signed_command( {"routable_message": routable_message}, ) - async def subscriptions( - self, device_token: str, device_type: str - ) -> dict[str, Any]: - """Returns the list of vehicles for which this mobile device currently subscribes to push notifications.""" - return await self._request( - Method.GET, - "api/1/subscriptions", - query={"device_token": device_token, "device_type": device_type}, - ) - - async def subscriptions_set( - self, device_token: str, device_type: str - ) -> dict[str, Any]: - """Allows a mobile device to specify which vehicles to receive push notifications from.""" - return await self._request( - Method.POST, - "api/1/subscriptions", - query={"device_token": device_token, "device_type": device_type}, - ) - async def vehicle(self, vehicle_tag: str | int) -> dict[str, Any]: """Returns information about a vehicle.""" return await self._request(Method.GET, f"api/1/vehicles/{vehicle_tag}") @@ -743,7 +729,7 @@ async def vehicle(self, vehicle_tag: str | int) -> dict[str, Any]: async def vehicle_data( self, vehicle_tag: str | int, - endpoints: List[VehicleDataEndpoint] | List[str] | None = None, + endpoints: List[VehicleDataEndpoint | str] | None = None, ) -> dict[str, Any]: """Makes a live call to the vehicle. This may return cached data if the vehicle is offline. For vehicles running firmware versions 2023.38+, location_data is required to fetch vehicle location. This will result in a location sharing icon to show on the vehicle UI.""" endpoint_payload = ";".join(endpoints) if endpoints else None @@ -753,37 +739,21 @@ async def vehicle_data( {"endpoints": endpoint_payload}, ) - async def vehicle_subscriptions( - self, device_token: str, device_type: DeviceType | str - ) -> dict[str, Any]: - """Returns the list of vehicles for which this mobile device currently subscribes to push notifications.""" - return await self._request( - Method.GET, - "api/1/vehicle_subscriptions", - {"device_token": device_token, "device_type": device_type}, - ) - - async def vehicle_subscriptions_set( - self, device_token: str, device_type: DeviceType | str - ) -> dict[str, Any]: - """Allows a mobile device to specify which vehicles to receive push notifications from.""" - return await self._request( - Method.POST, - "api/1/vehicle_subscriptions", - params={"device_token": device_token, "device_type": device_type}, - ) - async def wake_up(self, vehicle_tag: str | int) -> dict[str, Any]: """Wakes the vehicle from sleep, which is a state to minimize idle energy consumption.""" return await self._request(Method.POST, f"api/1/vehicles/{vehicle_tag}/wake_up") async def warranty_details(self, vin: str | None) -> dict[str, Any]: """Returns warranty details.""" - return await self._request(Method.GET, "api/1/dx/warranty/details", {vin: vin}) + return await self._request( + Method.GET, "api/1/dx/warranty/details", {"vin": vin} + ) async def fleet_status(self, vins: List[str]) -> dict[str, Any]: """Checks whether vehicles can accept Tesla commands protocol for the partner's public key""" - return await self._request(Method.GET, "api/1/vehicles/fleet_status", json=vins) + return await self._request( + Method.GET, "api/1/vehicles/fleet_status", json={"vins": vins} + ) async def fleet_telemetry_config_create( self, config: dict[str, Any] diff --git a/tesla_fleet_api/vehiclespecific.py b/tesla_fleet_api/vehiclespecific.py index 7f3633b..300bfce 100644 --- a/tesla_fleet_api/vehiclespecific.py +++ b/tesla_fleet_api/vehiclespecific.py @@ -1,4 +1,5 @@ -from typing import Any +from __future__ import annotations +from typing import Any, TYPE_CHECKING from .const import ( Trunk, ClimateKeeperMode, @@ -9,11 +10,17 @@ DeviceType, ) +if TYPE_CHECKING: + from .vehicle import Vehicle + class VehicleSpecific: """Class describing the Tesla Fleet API vehicle endpoints and commands for a specific vehicle.""" - def __init__(self, parent, vin: str | int | None = None): + _parent: Vehicle + vin: str + + def __init__(self, parent: Vehicle, vin: str): self._parent = parent self.vin = vin @@ -406,45 +413,17 @@ async def signed_command(self, routable_message: str) -> dict[str, Any]: """Signed Commands is a generic endpoint replacing legacy commands.""" return await self._parent.signed_command(self.vin, routable_message) - async def subscriptions( - self, device_token: str, device_type: str - ) -> dict[str, Any]: - """Returns the list of vehicles for which this mobile device currently subscribes to push notifications.""" - return await self._parent.subscriptions(self.vin, device_token, device_type) - - async def subscriptions_set( - self, device_token: str, device_type: str - ) -> dict[str, Any]: - """Allows a mobile device to specify which vehicles to receive push notifications from.""" - return await self._parent.subscriptions_set(self.vin, device_token, device_type) - async def vehicle(self) -> dict[str, Any]: """Returns information about a vehicle.""" return await self._parent.vehicle(self.vin) async def vehicle_data( self, - endpoints: list[VehicleDataEndpoint] | str | None = None, + endpoints: list[VehicleDataEndpoint | str] | None = None, ) -> dict[str, Any]: """Makes a live call to the vehicle. This may return cached data if the vehicle is offline. For vehicles running firmware versions 2023.38+, location_data is required to fetch vehicle location. This will result in a location sharing icon to show on the vehicle UI.""" return await self._parent.vehicle_data(self.vin, endpoints) - async def vehicle_subscriptions( - self, device_token: str, device_type: DeviceType | str - ) -> dict[str, Any]: - """Returns the list of vehicles for which this mobile device currently subscribes to push notifications.""" - return await self._parent.vehicle_subscriptions( - self.vin, device_token, device_type - ) - - async def vehicle_subscriptions_set( - self, device_token: str, device_type: DeviceType | str - ) -> dict[str, Any]: - """Allows a mobile device to specify which vehicles to receive push notifications from.""" - return await self._parent.vehicle_subscriptions_set( - self.vin, device_token, device_type - ) - async def wake_up(self) -> dict[str, Any]: """Wakes the vehicle from sleep, which is a state to minimize idle energy consumption.""" return await self._parent.wake_up(self.vin)