upgrade to python 3.10

.github/workflows/ci.yml

@@ -9,7 +9,7 @@ on:
       - main
 
 env:
-  DEFAULT_PYTHON: 3.9
+  DEFAULT_PYTHON: "3.10"
 
 jobs:
   code-quality:

.github/workflows/python-publish.yml

@@ -30,7 +30,7 @@ jobs:
       - name: Set up Python
         uses: actions/[email protected]
         with:
-          python-version: "3.9"
+          python-version: "3.10"
 
       - name: Install dependencies
         run: |

.pre-commit-config.yaml

@@ -5,12 +5,16 @@ ci:
     - pylint
     - verifyNoGetLogger
 
+default_language_version:
+  python: python3.10
+
 repos:
   - repo: https://github.com/asottile/pyupgrade
     rev: v2.34.0
     hooks:
       - id: pyupgrade
-        args: [--py39-plus]
+        args:
+          - --py310-plus
   - repo: https://github.com/psf/black
     rev: 22.3.0
     hooks:
@@ -26,7 +30,9 @@ repos:
           - --ignore-words-list=hass,deebot
           - --skip="./.*,*.csv,*.json"
           - --quiet-level=2
-        exclude_types: [csv, json]
+        exclude_types:
+          - csv
+          - json
   - repo: https://github.com/PyCQA/flake8
     rev: 4.0.1
     hooks:
@@ -72,13 +78,15 @@ repos:
         name: Check with mypy
         entry: scripts/run-in-env.sh mypy
         language: script
-        types: [python]
+        types:
+          - python
         require_serial: true
       - id: pylint
         name: Check with pylint
         entry: scripts/run-in-env.sh pylint
         language: script
-        types: [python]
+        types:
+          - python
         require_serial: true
         exclude: ^tests/.+
       - id: verifyNoGetLogger

deebot_client/_api_client.py

@@ -1,6 +1,6 @@
 """Internal api client module."""
 import asyncio
-from typing import Any, Optional
+from typing import Any
 from urllib.parse import urljoin
 
 from aiohttp import ClientResponseError
@@ -31,9 +31,9 @@ async def post(
         path: str,
         json: dict[str, Any],
         *,
-        query_params: Optional[dict[str, Any]] = None,
-        headers: Optional[dict[str, Any]] = None,
-        credentials: Optional[Credentials] = None,
+        query_params: dict[str, Any] | None = None,
+        headers: dict[str, Any] | None = None,
+        credentials: Credentials | None = None,
     ) -> dict[str, Any]:
         """Perform a post request."""
         url = _get_portal_url(self._config, path)

deebot_client/api_client.py

@@ -1,6 +1,6 @@
 """Api client module."""
 from datetime import datetime
-from typing import Any, Union
+from typing import Any
 from urllib.parse import urljoin
 
 from ._api_client import _InternalApiClient
@@ -80,7 +80,7 @@ async def get_product_iot_map(self) -> dict[str, Any]:
 
     async def send_command(
         self,
-        command: Union[Command, CustomCommand],
+        command: Command | CustomCommand,
         device_info: DeviceInfo,
     ) -> dict[str, Any]:
         """Send json command for given vacuum to the api."""

deebot_client/authentication.py

@@ -1,7 +1,7 @@
 """Authentication module."""
 import asyncio
 import time
-from typing import Any, Callable, Mapping, Optional, Union
+from typing import Any, Callable, Mapping
 
 from aiohttp import hdrs
 
@@ -117,7 +117,7 @@ async def __call_login_api(
         self, account_id: str, password_hash: str
     ) -> dict[str, Any]:
         _LOGGER.debug("calling login api")
-        params: dict[str, Union[str, int]] = {
+        params: dict[str, str | int] = {
             "account": account_id,
             "password": password_hash,
             "requestId": md5(str(time.time())),
@@ -136,12 +136,12 @@ async def __call_login_api(
 
     @staticmethod
     def __sign(
-        params: dict[str, Union[str, int]],
-        additional_sign_params: Mapping[str, Union[str, int]],
+        params: dict[str, str | int],
+        additional_sign_params: Mapping[str, str | int],
         key: str,
         secret: str,
-    ) -> dict[str, Union[str, int]]:
-        sign_data: dict[str, Union[str, int]] = {**additional_sign_params, **params}
+    ) -> dict[str, str | int]:
+        sign_data: dict[str, str | int] = {**additional_sign_params, **params}
         sign_on_text = (
             key
             + "".join([k + "=" + str(sign_data[k]) for k in sorted(sign_data.keys())])
@@ -153,7 +153,7 @@ def __sign(
 
     async def __call_auth_api(self, access_token: str, user_id: str) -> str:
         _LOGGER.debug("calling auth api")
-        params: dict[str, Union[str, int]] = {
+        params: dict[str, str | int] = {
             "uid": user_id,
             "accessToken": access_token,
             "bizType": "ECOVACS_IOT",
@@ -224,8 +224,8 @@ def __init__(
 
         self._lock = asyncio.Lock()
         self._on_credentials_changed: set[Callable[[Credentials], None]] = set()
-        self._credentials: Optional[Credentials] = None
-        self._refresh_task: Optional[asyncio.TimerHandle] = None
+        self._credentials: Credentials | None = None
+        self._refresh_task: asyncio.TimerHandle | None = None
 
     async def authenticate(self, force: bool = False) -> Credentials:
         """Authenticate on ecovacs servers."""

deebot_client/command.py

@@ -1,12 +1,12 @@
 """Base command."""
 from abc import ABC, abstractmethod
-from typing import Any, Union
+from typing import Any
 
 
 class Command(ABC):
     """Abstract command object."""
 
-    def __init__(self, args: Union[dict, list, None] = None) -> None:
+    def __init__(self, args: dict | list | None = None) -> None:
         if args is None:
             args = {}
         self._args = args
@@ -19,7 +19,7 @@ def name(cls) -> str:
         raise NotImplementedError
 
     @property
-    def args(self) -> Union[dict[str, Any], list]:
+    def args(self) -> dict[str, Any] | list:
         """Command additional arguments."""
         return self._args
 

deebot_client/commands/charge_state.py

@@ -1,5 +1,5 @@
 """Charge state commands."""
-from typing import Any, Optional
+from typing import Any
 
 from ..events import StatusEvent
 from ..message import HandlingResult
@@ -31,7 +31,7 @@ def _handle_body(cls, event_bus: EventBus, body: dict[str, Any]) -> HandlingResu
             # Call this also if code is not in the body
             return super()._handle_body(event_bus, body)
 
-        status: Optional[VacuumState] = None
+        status: VacuumState | None = None
         if body.get("msg", None) == "fail":
             if body["code"] == "30007":  # Already charging
                 status = VacuumState.DOCKED

deebot_client/commands/clean.py

@@ -1,6 +1,6 @@
 """Clean commands."""
 from enum import Enum, unique
-from typing import Any, Optional
+from typing import Any
 
 from ..events import StatusEvent
 from ..logging_filter import get_logger
@@ -69,7 +69,7 @@ def _handle_body_data_dict(
         :return: A message response
         """
 
-        status: Optional[VacuumState] = None
+        status: VacuumState | None = None
         state = data.get("state")
         if data.get("trigger") == "alert":
             status = VacuumState.ERROR

deebot_client/commands/clean_logs.py

@@ -1,5 +1,5 @@
 """clean log commands."""
-from typing import Any, Optional
+from typing import Any
 
 from ..events import CleanJobStatus, CleanLogEntry, CleanLogEvent
 from ..exceptions import DeebotError
@@ -26,7 +26,7 @@ def _handle_requested(
         :return: A message response
         """
         if response["ret"] == "ok":
-            resp_logs: Optional[list[dict]] = response.get("logs")
+            resp_logs: list[dict] | None = response.get("logs")
 
             # Ecovacs API is changing their API, this request may not work properly
             if resp_logs is not None and len(resp_logs) >= 0:

deebot_client/commands/common.py

@@ -1,7 +1,7 @@
 """Base commands."""
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
-from typing import Any, Mapping, Optional, Union, final
+from typing import Any, Mapping, final
 
 from ..command import Command
 from ..events import EnableEvent
@@ -17,7 +17,7 @@
 class CommandResult(HandlingResult):
     """Command result object."""
 
-    requested_commands: Optional[list[Command]] = None
+    requested_commands: list[Command] | None = None
 
     @classmethod
     def success(cls) -> "CommandResult":
@@ -130,7 +130,7 @@ class SetCommand(_ExecuteCommand, CommandWithMqttP2PHandling, ABC):
 
     def __init__(
         self,
-        args: Union[dict, list, None],
+        args: dict | list | None,
         **kwargs: Mapping[str, Any],
     ) -> None:
         if kwargs:
@@ -183,7 +183,7 @@ class SetEnableCommand(SetCommand):
     # required as name is class variable, will be overwritten in subclasses
     name = "__invalid__"
 
-    def __init__(self, enable: Union[int, bool], **kwargs: Mapping[str, Any]) -> None:
+    def __init__(self, enable: int | bool, **kwargs: Mapping[str, Any]) -> None:
         if isinstance(enable, bool):
             enable = 1 if enable else 0
         super().__init__({"enable": enable}, **kwargs)

deebot_client/commands/custom.py

@@ -1,5 +1,5 @@
 """Custom command module."""
-from typing import Any, Union
+from typing import Any
 
 from ..events import CustomCommandEvent
 from ..logging_filter import get_logger
@@ -12,7 +12,7 @@
 class CustomCommand:
     """Custom command, used when user wants to execute a command, which is not part of this library."""
 
-    def __init__(self, name: str, args: Union[dict, list, None] = None) -> None:
+    def __init__(self, name: str, args: dict | list | None = None) -> None:
         self._name = name
         if args is None:
             args = {}
@@ -24,7 +24,7 @@ def name(self) -> str:
         return self._name
 
     @property
-    def args(self) -> Union[dict[str, Any], list]:
+    def args(self) -> dict[str, Any] | list:
         """Command additional arguments."""
         return self._args
 

deebot_client/commands/fan_speed.py

@@ -1,5 +1,5 @@
 """(fan) speed commands."""
-from typing import Any, Mapping, Union
+from typing import Any, Mapping
 
 from ..events import FanSpeedEvent
 from ..message import HandlingResult
@@ -41,7 +41,7 @@ class SetFanSpeed(SetCommand):
     get_command = GetFanSpeed
 
     def __init__(
-        self, speed: Union[str, int, FanSpeedLevel], **kwargs: Mapping[str, Any]
+        self, speed: str | int | FanSpeedLevel, **kwargs: Mapping[str, Any]
     ) -> None:
         if isinstance(speed, str):
             speed = FanSpeedLevel.get(speed)

deebot_client/commands/life_span.py

@@ -1,5 +1,5 @@
 """Life span commands."""
-from typing import Any, Union
+from typing import Any
 
 from ..events import LifeSpan, LifeSpanEvent
 from ..message import HandlingResult, HandlingState
@@ -46,7 +46,7 @@ class ResetLifeSpan(_ExecuteCommand, CommandWithMqttP2PHandling):
     name = "resetLifeSpan"
 
     def __init__(
-        self, type: Union[str, LifeSpan]  # pylint: disable=redefined-builtin
+        self, type: str | LifeSpan  # pylint: disable=redefined-builtin
     ) -> None:
         if isinstance(type, LifeSpan):
             type = type.value

deebot_client/commands/map.py

@@ -1,5 +1,5 @@
 """Maps commands."""
-from typing import Any, Optional, Union
+from typing import Any
 
 from ..command import Command
 from ..events import (
@@ -104,9 +104,9 @@ class GetMapSet(CommandWithHandling):
     def __init__(
         self,
         mid: str,
-        type: Union[  # pylint: disable=redefined-builtin
-            MapSetType, str
-        ] = MapSetType.ROOMS,
+        type: (  # pylint: disable=redefined-builtin
+            MapSetType | str
+        ) = MapSetType.ROOMS,
     ) -> None:
         if isinstance(type, MapSetType):
             type = type.value
@@ -183,12 +183,10 @@ class GetMapSubSet(CommandWithHandling):
     def __init__(
         self,
         *,
-        mid: Union[str, int],
-        mssid: Union[str, int],
-        msid: Optional[Union[str, int]] = None,
-        type: Union[  # pylint: disable=redefined-builtin
-            MapSetType, str
-        ] = MapSetType.ROOMS
+        mid: str | int,
+        mssid: str | int,
+        msid: str | int | None = None,
+        type: (MapSetType | str) = MapSetType.ROOMS  # pylint: disable=redefined-builtin
     ) -> None:
         if isinstance(type, MapSetType):
             type = type.value

deebot_client/commands/water_info.py

@@ -1,5 +1,5 @@
 """Water info commands."""
-from typing import Any, Mapping, Union
+from typing import Any, Mapping
 
 from ..events import WaterAmount, WaterInfoEvent
 from ..message import HandlingResult
@@ -33,7 +33,7 @@ class SetWaterInfo(SetCommand):
     get_command = GetWaterInfo
 
     def __init__(
-        self, amount: Union[str, int, WaterAmount], **kwargs: Mapping[str, Any]
+        self, amount: str | int | WaterAmount, **kwargs: Mapping[str, Any]
     ) -> None:
         # removing "enable" as we don't can set it
         kwargs.pop("enable", None)