Add type annotations with Mypy checks.

.github/workflows/ci.yaml

@@ -3,8 +3,27 @@ name: Test tesla_api
 on: pull_request
 
 jobs:
-  build:
+  flake8:
+    name: Lint with flake8
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - uses: actions/setup-python@v2
+      - run: |
+          pip install -r requirements-flake8.txt
+          flake8
+
+  mypy:
+    name: Check annotations with Mypy
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - uses: actions/setup-python@v2
+      - run: pip install aiohttp mypy
+      - run: mypy
 
+  test:
+    name: Tests
     runs-on: ubuntu-latest
     strategy:
       matrix:
@@ -18,12 +37,8 @@ jobs:
         python-version: ${{ matrix.python-version }}
     - name: Install dependencies
       run: |
-        python -m pip install --upgrade pip
-        pip install -e .[dev]
-        pip install -r requirements-dev.txt
-    - name: Run flake8
-      run: |
-        flake8
+        python -m pip install -U pip
+        pip install aiohttp pytest-cov
     - name: Test with pytest
       run: |
         pytest -vv --cov=tesla_api --cov-report=xml

.gitignore

@@ -1,7 +1,8 @@
 test.py
 
+.mypy_cache/
 .vscode/
 **/__pycache__/
 build/
 tesla_api.egg-info/
-dist/
+dist/

.mypy.ini

@@ -0,0 +1,25 @@
+[mypy]
+files = tesla_api/
+check_untyped_defs = True
+follow_imports_for_stubs = True
+disallow_any_decorated = True
+disallow_any_expr = True
+disallow_any_explicit = True
+disallow_any_generics = True
+disallow_any_unimported = True
+disallow_incomplete_defs = True
+disallow_subclassing_any = True
+disallow_untyped_calls = True
+disallow_untyped_decorators = True
+disallow_untyped_defs = True
+implicit_reexport = False
+no_implicit_optional = True
+show_error_codes = True
+strict_equality = True
+warn_incomplete_stub = True
+warn_no_return = True
+warn_redundant_casts = True
+warn_unreachable = True
+warn_unused_configs = True
+warn_unused_ignores = True
+warn_return_any = True

requirements-dev.txt → requirements-flake8.txt

@@ -1,6 +1,5 @@
+flake8
 flake8-bugbear
 #flake8-docstrings  # TODO: Resolve violations.
 flake8-import-order
 flake8-quotes
-pytest
-pytest-cov

setup.py

@@ -12,6 +12,7 @@
     long_description=long_description,
     long_description_content_type="text/markdown",
     url="https://github.com/mlowijs/tesla_api",
+    package_data={"tesla_api": ["py.typed"]},
     packages=find_packages(),
     classifiers=[
         "Programming Language :: Python :: 3.7",

tesla_api/__init__.py

@@ -1,13 +1,22 @@
 import asyncio
 import json
 from datetime import datetime, timedelta
+from types import TracebackType
+from typing import (Awaitable, Callable, List, Literal, Mapping, Optional, Type, TypeVar,
+                    TypedDict, Union, cast)
 
 import aiohttp
 
+from .datatypes import (BaseResponse, EnergySite, ErrorResponse, ProductsResponse,
+                        TokenParams, TokenResponse, VehiclesResponse)
 from .energy import Energy
-from .exceptions import ApiError, AuthenticationError, VehicleUnavailableError
+from .exceptions import (ApiError, AuthenticationError, VehicleInServiceError,
+                         VehicleUnavailableError)
 from .vehicle import Vehicle
 
+__all__ = ("Energy", "Vehicle", "TeslaApiClient", "ApiError", "AuthenticationError",
+           "VehicleInServiceError", "VehicleUnavailableError")
+
 TESLA_API_BASE_URL = "https://owner-api.teslamotors.com/"
 TOKEN_URL = TESLA_API_BASE_URL + "oauth/token"
 API_URL = TESLA_API_BASE_URL + "api/1"
@@ -16,12 +25,35 @@
 OAUTH_CLIENT_SECRET = "c7257eb71a564034f9419ee651c7d0e5f7aa6bfbd18bafb5c5c033b093bb2fa3"
 
 
+class AuthHeaders(TypedDict):
+    Authorization: str
+
+
+class _AuthParamsPassword(TypedDict):
+    grant_type: Literal["password"]
+    email: str
+    password: str
+
+
+class _AuthParamsRefresh(TypedDict):
+    grant_type: Literal["refresh_token"]
+    refresh_token: str
+
+
+AuthParams = Union[_AuthParamsPassword, _AuthParamsRefresh]
+T = TypeVar("T", bound="TeslaApiClient")
+
+
 class TeslaApiClient:
-    callback_update = None  # Called when vehicle's state has been updated.
-    callback_wake_up = None  # Called when attempting to wake a vehicle.
+    # Called when vehicle's state has been updated.
+    callback_update: Optional[Callable[[Vehicle], Awaitable[None]]] = None
+    # Called when attempting to wake a vehicle.
+    callback_wake_up: Optional[Callable[[Vehicle], Awaitable[None]]] = None
     timeout = 30  # Default timeout for operations such as Vehicle.wake_up().
 
-    def __init__(self, email=None, password=None, token=None, on_new_token=None):
+    def __init__(self, email: Optional[str] = None, password: Optional[str] = None,
+                 token: Optional[str] = None,
+                 on_new_token: Optional[Callable[[str], Awaitable[None]]] = None):
         """Creates client from provided credentials.
 
         If token is not provided, or is no longer valid, then a new token will
@@ -35,28 +67,22 @@ def __init__(self, email=None, password=None, token=None, on_new_token=None):
         assert token is not None or (email is not None and password is not None)
         self._email = email
         self._password = password
-        self._token = json.loads(token) if token else None
+        self._token = cast(TokenResponse, json.loads(token)) if token else None
         self._new_token_callback = on_new_token
         self._session = aiohttp.ClientSession()
 
-    async def __aenter__(self):
-        return self
-
-    async def __aexit__(self, exc_type, exc_val, exc_tb):
-        await self.close()
-
-    async def close(self):
+    async def close(self) -> None:
         await self._session.close()
 
-    async def _get_token(self, data):
-        request_data = {
+    async def _get_token(self, data: AuthParams) -> TokenResponse:
+        request_data = cast(TokenParams, {
             "client_id": OAUTH_CLIENT_ID,
             "client_secret": OAUTH_CLIENT_SECRET,
-        }
-        request_data.update(data)
+            **data
+        })
 
         async with self._session.post(TOKEN_URL, data=request_data) as resp:
-            response_json = await resp.json()
+            response_json = await cast(Awaitable[TokenResponse], resp.json())
             if resp.status == 401:
                 raise AuthenticationError(response_json)
 
@@ -66,15 +92,17 @@ async def _get_token(self, data):
 
         return response_json
 
-    async def _get_new_token(self):
-        data = {"grant_type": "password", "email": self._email, "password": self._password}
+    async def _get_new_token(self) -> TokenResponse:
+        assert self._email is not None and self._password is not None
+        data: _AuthParamsPassword = {"grant_type": "password", "email": self._email,
+                                     "password": self._password}
         return await self._get_token(data)
 
-    async def _refresh_token(self, refresh_token):
-        data = {"grant_type": "refresh_token", "refresh_token": refresh_token}
+    async def _refresh_token(self, refresh_token: str) -> TokenResponse:
+        data: _AuthParamsRefresh = {"grant_type": "refresh_token", "refresh_token": refresh_token}
         return await self._get_token(data)
 
-    async def authenticate(self):
+    async def authenticate(self) -> None:
         if not self._token:
             self._token = await self._get_new_token()
 
@@ -84,40 +112,53 @@ async def authenticate(self):
         if datetime.utcnow() >= expiration_date:
             self._token = await self._refresh_token(self._token["refresh_token"])
 
-    def _get_headers(self):
-        return {"Authorization": "Bearer {}".format(self._token["access_token"])}
-
-    async def get(self, endpoint, params=None):
-        await self.authenticate()
-        url = "{}/{}".format(API_URL, endpoint)
+    def _get_headers(self) -> AuthHeaders:
+        assert self._token is not None
+        return {
+            "Authorization": "Bearer {}".format(self._token["access_token"])
+        }
 
-        async with self._session.get(url, headers=self._get_headers(), params=params) as resp:
-            response_json = await resp.json()
+    async def get(self, endpoint: str, params: Optional[Mapping[str, str]] = None) -> object:
+        return await self._send_request("get", endpoint, params=params)
 
-        if "error" in response_json:
-            if "vehicle unavailable" in response_json["error"]:
-                raise VehicleUnavailableError()
-            raise ApiError(response_json["error"])
+    async def post(self, endpoint: str, data: Optional[Mapping[str, object]] = None) -> object:
+        return await self._send_request("post", endpoint, data=data)
 
-        return response_json["response"]
-
-    async def post(self, endpoint, data=None):
+    async def _send_request(self, method: Literal["get", "post"], endpoint: str, *,
+                            data: Optional[Mapping[str, object]] = None,
+                            params: Optional[Mapping[str, str]] = None) -> object:
         await self.authenticate()
         url = "{}/{}".format(API_URL, endpoint)
 
-        async with self._session.post(url, headers=self._get_headers(), json=data) as resp:
-            response_json = await resp.json()
+        async with self._session.request(method, url, headers=self._get_headers(),
+                                         json=data, params=params) as resp:
+            # TODO(Mypy): https://github.com/python/mypy/issues/8884
+            response_json = await cast(Awaitable[Union[BaseResponse, ErrorResponse]], resp.json())
 
         if "error" in response_json:
-            if "vehicle unavailable" in response_json["error"]:
+            error_response = cast(ErrorResponse, response_json)
+            error = error_response["error"]
+            if "vehicle unavailable" in error:
                 raise VehicleUnavailableError()
-            raise ApiError(response_json["error"])
+            elif "in service" in error:
+                raise VehicleInServiceError()
+            raise ApiError(error)
 
+        response_json = cast(BaseResponse, response_json)
         return response_json["response"]
 
-    async def list_vehicles(self):
-        return [Vehicle(self, vehicle) for vehicle in await self.get("vehicles")]
+    async def list_vehicles(self) -> List[Vehicle]:
+        vehicles = cast(VehiclesResponse, await self.get("vehicles"))
+        return [Vehicle(self, v) for v in vehicles]
+
+    async def list_energy_sites(self) -> List[Energy]:
+        products = cast(ProductsResponse, await self.get("products"))
+        return [Energy(self, cast(EnergySite, p)["energy_site_id"])
+                for p in products if "energy_site_id" in p]
 
-    async def list_energy_sites(self):
-        return [Energy(self, product["energy_site_id"])
-                for product in await self.get("products") if "energy_site_id" in product]
+    async def __aenter__(self: T) -> T:
+        return self
+
+    async def __aexit__(self, exc_type: Optional[Type[BaseException]],
+                        exc_val: Optional[BaseException], exc_tb: Optional[TracebackType]) -> None:
+        await self.close()

tesla_api/charge.py

@@ -1,22 +1,31 @@
+from typing import TYPE_CHECKING, cast
+
+from .datatypes import ChargeStateResponse
+
+if TYPE_CHECKING:
+    from .vehicle import Vehicle
+
+
 class Charge:
-    def __init__(self, vehicle):
+    def __init__(self, vehicle: "Vehicle"):
         self._vehicle = vehicle
         self._api_client = vehicle._api_client
 
-    async def get_state(self):
-        return await self._api_client.get(
-            "vehicles/{}/data_request/charge_state".format(self._vehicle.id))
+    async def get_state(self) -> ChargeStateResponse:
+        endpoint = "vehicles/{}/data_request/charge_state".format(self._vehicle.id)
+        return cast(ChargeStateResponse, await self._api_client.get(endpoint))
 
-    async def start_charging(self):
-        return await self._vehicle._command("charge_start")
+    async def start_charging(self) -> bool:
+        return cast(bool, await self._vehicle._command("charge_start"))
 
-    async def stop_charging(self):
-        return await self._vehicle._command("charge_stop")
+    async def stop_charging(self) -> bool:
+        return cast(bool, await self._vehicle._command("charge_stop"))
 
-    async def set_charge_limit(self, percentage):
+    async def set_charge_limit(self, percentage: int) -> bool:  # TODO: int or float?
         percentage = round(percentage)
 
         if not (50 <= percentage <= 100):
             raise ValueError("Percentage should be between 50 and 100")
 
-        return await self._vehicle._command("set_charge_limit", {"percent": percentage})
+        args = {"percent": percentage}
+        return cast(bool, await self._vehicle._command("set_charge_limit", args))