typing: modernize annotations

rnovatorov · rnovatorov · commit 76cc5f0835c8 · 2025-10-28T19:04:45.000+01:00
diff --git a/src/enapter/http/api/config.py b/src/enapter/http/api/config.py
@@ -1,5 +1,5 @@
 import os
-from typing import MutableMapping, Optional
+from typing import MutableMapping
 
 
 class Config:
@@ -16,7 +16,7 @@ def from_env(
             base_url=env.get(prefix + "BASE_URL"),
         )
 
-    def __init__(self, token: str, base_url: Optional[str] = None) -> None:
+    def __init__(self, token: str, base_url: str | None = None) -> None:
         if not token:
             raise ValueError("token is missing")
         self.token = token
diff --git a/src/enapter/http/api/devices/communication_config.py b/src/enapter/http/api/devices/communication_config.py
@@ -1,5 +1,5 @@
 import dataclasses
-from typing import Any, Dict
+from typing import Any
 
 from .mqtt_credentials import MQTTCredentials
 from .mqtt_protocol import MQTTProtocol
@@ -21,7 +21,7 @@ class CommunicationConfig:
     channel_id: str
 
     @classmethod
-    def from_dto(cls, dto: Dict[str, Any]) -> "CommunicationConfig":
+    def from_dto(cls, dto: dict[str, Any]) -> "CommunicationConfig":
         mqtt_protocol = MQTTProtocol(dto["mqtt_protocol"])
         mqtt_credentials: MQTTCredentials | MQTTSCredentials | None = None
         match mqtt_protocol:
diff --git a/src/enapter/http/api/devices/device.py b/src/enapter/http/api/devices/device.py
@@ -1,6 +1,6 @@
 import dataclasses
 import datetime
-from typing import Any, Dict
+from typing import Any
 
 from .authorized_role import AuthorizedRole
 from .device_type import DeviceType
@@ -19,7 +19,7 @@ class Device:
     authorized_role: AuthorizedRole
 
     @classmethod
-    def from_dto(cls, dto: Dict[str, Any]) -> "Device":
+    def from_dto(cls, dto: dict[str, Any]) -> "Device":
         return cls(
             id=dto["id"],
             blueprint_id=dto["blueprint_id"],
diff --git a/src/enapter/http/api/devices/mqtt_credentials.py b/src/enapter/http/api/devices/mqtt_credentials.py
@@ -1,5 +1,5 @@
 import dataclasses
-from typing import Any, Dict
+from typing import Any
 
 
 @dataclasses.dataclass
@@ -9,5 +9,5 @@ class MQTTCredentials:
     password: str
 
     @classmethod
-    def from_dto(cls, dto: Dict[str, Any]) -> "MQTTCredentials":
+    def from_dto(cls, dto: dict[str, Any]) -> "MQTTCredentials":
         return cls(username=dto["username"], password=dto["password"])
diff --git a/src/enapter/http/api/devices/mqtts_credentials.py b/src/enapter/http/api/devices/mqtts_credentials.py
@@ -1,5 +1,5 @@
 import dataclasses
-from typing import Any, Dict
+from typing import Any
 
 
 @dataclasses.dataclass
@@ -10,7 +10,7 @@ class MQTTSCredentials:
     ca_chain: str
 
     @classmethod
-    def from_dto(cls, dto: Dict[str, Any]) -> "MQTTSCredentials":
+    def from_dto(cls, dto: dict[str, Any]) -> "MQTTSCredentials":
         return cls(
             private_key=dto["private_key"],
             certificate=dto["certificate"],
diff --git a/src/enapter/log/json_formatter.py b/src/enapter/log/json_formatter.py
@@ -1,17 +1,15 @@
 import datetime
 import logging
-from typing import Any, Dict
+from typing import Any
 
 import json_log_formatter  # type: ignore
 
 
 class JSONFormatter(json_log_formatter.JSONFormatter):
+
     def json_record(
-        self,
-        message: str,
-        extra: Dict[str, Any],
-        record: logging.LogRecord,
-    ) -> Dict[str, Any]:
+        self, message: str, extra: dict[str, Any], record: logging.LogRecord
+    ) -> dict[str, Any]:
         try:
             del extra["taskName"]
         except KeyError:
@@ -34,5 +32,5 @@ def json_record(
 
         return json_record
 
-    def mutate_json_record(self, json_record: Dict[str, Any]) -> Dict[str, Any]:
+    def mutate_json_record(self, json_record: dict[str, Any]) -> dict[str, Any]:
         return json_record
diff --git a/src/enapter/mqtt/api/command_request.py b/src/enapter/mqtt/api/command_request.py
@@ -1,5 +1,5 @@
 import dataclasses
-from typing import Any, Dict
+from typing import Any
 
 from .command_response import CommandResponse
 from .command_state import CommandState
@@ -11,25 +11,25 @@ class CommandRequest(Message):
 
     id: str
     name: str
-    arguments: Dict[str, Any]
+    arguments: dict[str, Any]
 
     @classmethod
-    def from_dto(cls, dto: Dict[str, Any]) -> "CommandRequest":
+    def from_dto(cls, dto: dict[str, Any]) -> "CommandRequest":
         return cls(
             id=dto["id"],
             name=dto["name"],
             arguments=dto.get("arguments", {}),
         )
 
-    def to_dto(self) -> Dict[str, Any]:
+    def to_dto(self) -> dict[str, Any]:
         return {
             "id": self.id,
             "name": self.name,
             "arguments": self.arguments,
         }
 
     def new_response(
-        self, state: CommandState, payload: Dict[str, Any]
+        self, state: CommandState, payload: dict[str, Any]
     ) -> CommandResponse:
         return CommandResponse(
             id=self.id,
diff --git a/src/enapter/mqtt/api/command_response.py b/src/enapter/mqtt/api/command_response.py
@@ -1,5 +1,5 @@
 import dataclasses
-from typing import Any, Dict
+from typing import Any
 
 from .command_state import CommandState
 from .message import Message
@@ -10,17 +10,17 @@ class CommandResponse(Message):
 
     id: str
     state: CommandState
-    payload: Dict[str, Any]
+    payload: dict[str, Any]
 
     @classmethod
-    def from_dto(cls, dto: Dict[str, Any]) -> "CommandResponse":
+    def from_dto(cls, dto: dict[str, Any]) -> "CommandResponse":
         return cls(
             id=dto["id"],
             state=CommandState(dto["state"]),
             payload=dto.get("payload", {}),
         )
 
-    def to_dto(self) -> Dict[str, Any]:
+    def to_dto(self) -> dict[str, Any]:
         return {
             "id": self.id,
             "state": self.state.value,
diff --git a/src/enapter/mqtt/api/log.py b/src/enapter/mqtt/api/log.py
@@ -1,5 +1,5 @@
 import dataclasses
-from typing import Any, Dict
+from typing import Any
 
 from .log_severity import LogSeverity
 from .message import Message
@@ -14,15 +14,15 @@ class Log(Message):
     persist: bool
 
     @classmethod
-    def from_dto(cls, dto: Dict[str, Any]) -> "Log":
+    def from_dto(cls, dto: dict[str, Any]) -> "Log":
         return cls(
             timestamp=dto["timestamp"],
             message=dto["message"],
             severity=LogSeverity(dto["severity"]),
             persist=dto["persist"],
         )
 
-    def to_dto(self) -> Dict[str, Any]:
+    def to_dto(self) -> dict[str, Any]:
         return {
             "timestamp": self.timestamp,
             "message": self.message,
diff --git a/src/enapter/mqtt/api/message.py b/src/enapter/mqtt/api/message.py
@@ -1,21 +1,21 @@
 import abc
 import json
-from typing import Any, Dict, Union
+from typing import Any
 
 
 class Message(abc.ABC):
 
     @classmethod
     @abc.abstractmethod
-    def from_dto(cls, dto: Dict[str, Any]):
+    def from_dto(cls, dto: dict[str, Any]):
         pass
 
     @abc.abstractmethod
-    def to_dto(self) -> Dict[str, Any]:
+    def to_dto(self) -> dict[str, Any]:
         pass
 
     @classmethod
-    def from_json(cls, data: Union[str, bytes]):
+    def from_json(cls, data: str | bytes):
         dto = json.loads(data)
         return cls.from_dto(dto)
 
diff --git a/src/enapter/mqtt/api/properties.py b/src/enapter/mqtt/api/properties.py
@@ -1,5 +1,5 @@
 import dataclasses
-from typing import Any, Dict
+from typing import Any
 
 from .message import Message
 
@@ -8,17 +8,17 @@
 class Properties(Message):
 
     timestamp: int
-    values: Dict[str, Any] = dataclasses.field(default_factory=dict)
+    values: dict[str, Any] = dataclasses.field(default_factory=dict)
 
     def __post_init__(self) -> None:
         if "timestamp" in self.values:
             raise ValueError("`timestamp` is reserved")
 
     @classmethod
-    def from_dto(cls, dto: Dict[str, Any]) -> "Properties":
+    def from_dto(cls, dto: dict[str, Any]) -> "Properties":
         timestamp = dto["timestamp"]
         values = {k: v for k, v in dto.items() if k != "timestamp"}
         return cls(timestamp=timestamp, values=values)
 
-    def to_dto(self) -> Dict[str, Any]:
+    def to_dto(self) -> dict[str, Any]:
         return {"timestamp": self.timestamp, **self.values}
diff --git a/src/enapter/mqtt/api/telemetry.py b/src/enapter/mqtt/api/telemetry.py
@@ -1,5 +1,5 @@
 import dataclasses
-from typing import Any, Dict, List, Optional
+from typing import Any
 
 from .message import Message
 
@@ -8,8 +8,8 @@
 class Telemetry(Message):
 
     timestamp: int
-    alerts: Optional[List[str]] = None
-    values: Dict[str, Any] = dataclasses.field(default_factory=dict)
+    alerts: list[str] | None = None
+    values: dict[str, Any] = dataclasses.field(default_factory=dict)
 
     def __post_init__(self) -> None:
         if "timestamp" in self.values:
@@ -18,13 +18,13 @@ def __post_init__(self) -> None:
             raise ValueError("`alerts` is reserved")
 
     @classmethod
-    def from_dto(cls, dto: Dict[str, Any]) -> "Telemetry":
+    def from_dto(cls, dto: dict[str, Any]) -> "Telemetry":
         dto = dto.copy()
         timestamp = dto.pop("timestamp")
         alerts = dto.pop("alerts", None)
         return cls(timestamp=timestamp, alerts=alerts, values=dto)
 
-    def to_dto(self) -> Dict[str, Any]:
+    def to_dto(self) -> dict[str, Any]:
         dto = {"timestamp": self.timestamp, **self.values}
         if self.alerts is not None:
             dto["alerts"] = self.alerts
diff --git a/src/enapter/mqtt/client.py b/src/enapter/mqtt/client.py
@@ -3,7 +3,7 @@
 import logging
 import ssl
 import tempfile
-from typing import AsyncGenerator, Optional
+from typing import AsyncGenerator
 
 import aiomqtt  # type: ignore
 
@@ -22,7 +22,7 @@ def __init__(self, task_group: asyncio.TaskGroup, config: Config) -> None:
         self._config = config
         self._mdns_resolver = mdns.Resolver()
         self._tls_context = self._new_tls_context(config)
-        self._publisher: Optional[aiomqtt.Client] = None
+        self._publisher: aiomqtt.Client | None = None
         self._publisher_connected = asyncio.Event()
         self._task = task_group.create_task(self._run())
 
@@ -89,7 +89,7 @@ async def _connect(self) -> AsyncGenerator[aiomqtt.Client, None]:
             yield client
 
     @staticmethod
-    def _new_tls_context(config: Config) -> Optional[ssl.SSLContext]:
+    def _new_tls_context(config: Config) -> ssl.SSLContext | None:
         if config.tls is None:
             return None
 
diff --git a/src/enapter/mqtt/config.py b/src/enapter/mqtt/config.py
@@ -1,13 +1,13 @@
 import os
-from typing import MutableMapping, Optional
+from typing import MutableMapping
 
 
 class TLSConfig:
 
     @classmethod
     def from_env(
         cls, env: MutableMapping[str, str] = os.environ, namespace: str = "ENAPTER_"
-    ) -> Optional["TLSConfig"]:
+    ) -> "TLSConfig" | None:
         prefix = namespace + "MQTT_TLS_"
 
         secret_key = env.get(prefix + "SECRET_KEY")
@@ -55,9 +55,9 @@ def __init__(
         self,
         host: str,
         port: int,
-        user: Optional[str] = None,
-        password: Optional[str] = None,
-        tls_config: Optional[TLSConfig] = None,
+        user: str | None = None,
+        password: str | None = None,
+        tls_config: TLSConfig | None = None,
     ) -> None:
         self.host = host
         self.port = port
diff --git a/src/enapter/standalone/config.py b/src/enapter/standalone/config.py
@@ -3,7 +3,7 @@
 import enum
 import json
 import os
-from typing import Any, Dict, MutableMapping, Self
+from typing import Any, MutableMapping, Self
 
 from enapter import mqtt
 
@@ -123,7 +123,7 @@ class CommunicationConfigV1:
     channel_id: str
 
     @classmethod
-    def from_dto(cls, dto: Dict[str, Any]) -> Self:
+    def from_dto(cls, dto: dict[str, Any]) -> Self:
         return cls(
             mqtt_host=dto["mqtt_host"],
             mqtt_port=int(dto["mqtt_port"]),
@@ -150,7 +150,7 @@ class MQTTCredentials:
         password: str
 
         @classmethod
-        def from_dto(cls, dto: Dict[str, Any]) -> Self:
+        def from_dto(cls, dto: dict[str, Any]) -> Self:
             return cls(username=dto["username"], password=dto["password"])
 
     @dataclasses.dataclass
@@ -161,7 +161,7 @@ class MQTTSCredentials:
         ca_chain: str
 
         @classmethod
-        def from_dto(cls, dto: Dict[str, Any]) -> Self:
+        def from_dto(cls, dto: dict[str, Any]) -> Self:
             return cls(
                 private_key=dto["private_key"],
                 certificate=dto["certificate"],
@@ -176,7 +176,7 @@ def from_dto(cls, dto: Dict[str, Any]) -> Self:
     channel_id: str
 
     @classmethod
-    def from_dto(cls, dto: Dict[str, Any]) -> Self:
+    def from_dto(cls, dto: dict[str, Any]) -> Self:
         mqtt_protocol = cls.MQTTProtocol(dto["mqtt_protocol"])
         mqtt_credentials: (
             CommunicationConfigV3.MQTTCredentials
diff --git a/src/enapter/standalone/device.py b/src/enapter/standalone/device.py
diff --git a/src/enapter/standalone/logger.py b/src/enapter/standalone/logger.py
