Skip to content

Commit

Permalink
Make sub-classes optional
Browse files Browse the repository at this point in the history
  • Loading branch information
Bre77 committed Jan 7, 2024
1 parent ddd820a commit 4e34edc
Show file tree
Hide file tree
Showing 4 changed files with 56 additions and 40 deletions.
42 changes: 21 additions & 21 deletions tesla_fleet_api/exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -211,16 +211,16 @@ async def raise_for_status(resp: aiohttp.ClientResponse) -> None:
"""Raise an exception if the response status code is >=400."""
# https://developer.tesla.com/docs/fleet-api#response-codes

if e.status == 401 and resp.content_length == 0:
if resp.status == 401 and resp.content_length == 0:
# This error does not return a body
raise OAuthExpired() from e
raise OAuthExpired()

data = await resp.json()

try:
resp.raise_for_status()
except aiohttp.ClientResponseError as e:
if e.status == 400:
if resp.status == 400:
if data.error == Errors.INVALID_COMMAND:
raise InvalidCommand(data) from e
elif data.error == Errors.INVALID_FIELD:
Expand All @@ -233,44 +233,44 @@ async def raise_for_status(resp: aiohttp.ClientResponse) -> None:
raise InvalidRedirectUrl(data) from e
elif data.error == Errors.UNAUTHORIZED_CLIENT:
raise UnauthorizedClient(data) from e
elif e.status == 401:
elif resp.status == 401:
if data.error == Errors.MOBILE_ACCESS_DISABLED:
raise MobileAccessDisabled(data) from e
elif data.error == Errors.INVALID_TOKEN:
raise InvalidToken(data) from e
elif e.status == 402:
elif resp.status == 402:
raise PaymentRequired(data) from e
elif e.status == 403:
elif resp.status == 403:
raise Forbidden(data) from e
elif e.status == 404:
elif resp.status == 404:
raise NotFound(data) from e
elif e.status == 405:
elif resp.status == 405:
raise NotAllowed(data) from e
elif e.status == 406:
elif resp.status == 406:
raise NotAcceptable(data) from e
elif e.status == 408:
elif resp.status == 408:
raise VehicleOffline(data) from e
elif e.status == 412:
elif resp.status == 412:
raise PreconditionFailed(data) from e
elif e.status == 421:
elif resp.status == 421:
raise InvalidRegion(data) from e
elif e.status == 422:
elif resp.status == 422:
raise InvalidResource(data) from e
elif e.status == 423:
elif resp.status == 423:
raise Locked(data) from e
elif e.status == 429:
elif resp.status == 429:
raise RateLimited(data) from e
elif e.status == 451:
elif resp.status == 451:
raise ResourceUnavailableForLegalReasons(data) from e
elif e.status == 499:
elif resp.status == 499:
raise ClientClosedRequest(data) from e
elif e.status == 500:
elif resp.status == 500:
raise InternalServerError(data) from e
elif e.status == 503:
elif resp.status == 503:
raise ServiceUnavailable(data) from e
elif e.status == 504:
elif resp.status == 504:
raise GatewayTimeout(data) from e
elif e.status == 540:
elif resp.status == 540:
raise DeviceUnexpectedResponse(data) from e
else:
raise e
22 changes: 17 additions & 5 deletions tesla_fleet_api/teslafleetapi.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
from .partner import Partner
from .user import User
from .vehicle import Vehicle
from .vehiclespecific import VehicleSpecific


# Based on https://developer.tesla.com/docs/fleet-api
Expand All @@ -20,6 +21,7 @@ class TeslaFleetApi:
session: aiohttp.ClientSession
headers: dict[str, str]
raise_for_status: bool
vehicles: list[VehicleSpecific]

def __init__(
self,
Expand All @@ -29,6 +31,11 @@ def __init__(
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,
user_scope: bool = True,
vehicle_scope: bool = True,
):
"""Initialize the Tesla Fleet API."""

Expand All @@ -41,11 +48,16 @@ def __init__(
self.server = server or SERVERS.get(region)
self.raise_for_status = raise_for_status

self.charging = Charging(self)
self.energy = Energy(self)
self.user = User(self)
self.partner = Partner(self)
self.vehicle = Vehicle(self)
if charging_scope:
self.charging = Charging(self)
if energy_scope:
self.energy = Energy(self)
if user_scope:
self.user = User(self)
if partner_scope:
self.partner = Partner(self)
if vehicle_scope:
self.vehicle = Vehicle(self)

async def find_server(self) -> None:
"""Find the server URL for the Tesla Fleet API."""
Expand Down
27 changes: 15 additions & 12 deletions tesla_fleet_api/teslemetry.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,12 +5,22 @@
from .const import Methods


class TeslemetryVehicle(Vehicle):
"""Tesla Fleet API Vehicle."""

async def create(self, only_subscribed=True) -> [VehicleSpecific]:
"""Creates a class for each vehicle."""
if only_subscribed:
return [VehicleSpecific(self, vin) for vin in await self._parent.vehicles()]
return await super().create()


class Teslemetry(TeslaFleetApi):
def __init__(
self,
session: aiohttp.ClientSession,
access_token: str,
raise_for_status: bool = False,
raise_for_status: bool = True,
):
"""Initialize the Teslemetry API."""
super().__init__(
Expand All @@ -19,7 +29,11 @@ def __init__(
use_command_protocol=False,
server="https://teslemetry.com",
raise_for_status=raise_for_status,
partner_scope=False,
user_scope=False,
vehicle_scope=False,
)
self.vehicle = TeslemetryVehicle(self)

async def vehicles(self):
"""Get the subscribed vehicles."""
Expand All @@ -31,14 +45,3 @@ async def vehicles(self):
async def find_server(self):
"""Find the server URL for the Tesla Fleet API."""
raise NotImplementedError("Do not use this function for Teslemetry.")

class Vehicle(Vehicle):
"""Tesla Fleet API Vehicle."""

async def create(self, only_subscribed=True) -> [VehicleSpecific]:
"""Creates a class for each vehicle."""
if only_subscribed:
return [
VehicleSpecific(self, vin) for vin in await self._parent.vehicles()
]
return await super().create()
5 changes: 3 additions & 2 deletions tesla_fleet_api/vehicle.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,8 +22,9 @@ def __init__(self, parent):

async def create(self) -> [VehicleSpecific]:
"""Creates a class for each vehicle."""
if vehicles := await self.list()["response"]:
return [VehicleSpecific(self, x["vin"]) for x in vehicles]
if vehicles := (await self.list()).get("response"):
self._parent.vehicles = [VehicleSpecific(self, x["vin"]) for x in vehicles]
return self._parent.vehicles
return []

async def actuate_trunk(
Expand Down

0 comments on commit 4e34edc

Please sign in to comment.