Enable strict typing for apache_kafka (#106823)

.strict-typing

@@ -75,6 +75,7 @@ homeassistant.components.androidtv_remote.*
 homeassistant.components.anel_pwrctrl.*
 homeassistant.components.anova.*
 homeassistant.components.anthemav.*
+homeassistant.components.apache_kafka.*
 homeassistant.components.apcupsd.*
 homeassistant.components.apprise.*
 homeassistant.components.aprs.*

homeassistant/components/apache_kafka/__init__.py

@@ -1,7 +1,10 @@
 """Support for Apache Kafka."""
+from __future__ import annotations
+
 from datetime import datetime
 import json
 import sys
+from typing import Any, Literal
 
 import voluptuous as vol
 
@@ -15,11 +18,12 @@ from homeassistant.const import (
     STATE_UNAVAILABLE,
     STATE_UNKNOWN,
 )
-from homeassistant.core import HomeAssistant
+from homeassistant.core import Event, HomeAssistant
 from homeassistant.exceptions import HomeAssistantError
 import homeassistant.helpers.config_validation as cv
-from homeassistant.helpers.entityfilter import FILTER_SCHEMA
+from homeassistant.helpers.entityfilter import FILTER_SCHEMA, EntityFilter
-from homeassistant.helpers.typing import ConfigType
+from homeassistant.helpers.event import EventStateChangedData
+from homeassistant.helpers.typing import ConfigType, EventType
 from homeassistant.util import ssl as ssl_util
 
 if sys.version_info < (3, 12):
@@ -84,11 +88,11 @@ class DateTimeJSONEncoder(json.JSONEncoder):
     Additionally add encoding for datetime objects as isoformat.
     """
 
-    def default(self, o):
+    def default(self, o: Any) -> str:
         """Implement encoding logic."""
         if isinstance(o, datetime):
             return o.isoformat()
-        return super().default(o)
+        return super().default(o)  # type: ignore[no-any-return]
 
 
 class KafkaManager:
@@ -96,15 +100,15 @@ class KafkaManager:
 
     def __init__(
         self,
-        hass,
+        hass: HomeAssistant,
-        ip_address,
+        ip_address: str,
-        port,
+        port: int,
-        topic,
+        topic: str,
-        entities_filter,
+        entities_filter: EntityFilter,
-        security_protocol,
+        security_protocol: Literal["PLAINTEXT", "SASL_SSL"],
-        username,
+        username: str | None,
-        password,
+        password: str | None,
-    ):
+    ) -> None:
         """Initialize."""
         self._encoder = DateTimeJSONEncoder()
         self._entities_filter = entities_filter
@@ -121,30 +125,30 @@ class KafkaManager:
         )
         self._topic = topic
 
-    def _encode_event(self, event):
+    def _encode_event(self, event: EventType[EventStateChangedData]) -> bytes | None:
         """Translate events into a binary JSON payload."""
-        state = event.data.get("new_state")
+        state = event.data["new_state"]
         if (
             state is None
             or state.state in (STATE_UNKNOWN, "", STATE_UNAVAILABLE)
             or not self._entities_filter(state.entity_id)
         ):
-            return
+            return None
 
         return json.dumps(obj=state.as_dict(), default=self._encoder.encode).encode(
             "utf-8"
         )
 
-    async def start(self):
+    async def start(self) -> None:
         """Start the Kafka manager."""
-        self._hass.bus.async_listen(EVENT_STATE_CHANGED, self.write)
+        self._hass.bus.async_listen(EVENT_STATE_CHANGED, self.write)  # type: ignore[arg-type]
         await self._producer.start()
 
-    async def shutdown(self, _):
+    async def shutdown(self, _: Event) -> None:
         """Shut the manager down."""
         await self._producer.stop()
 
-    async def write(self, event):
+    async def write(self, event: EventType[EventStateChangedData]) -> None:
         """Write a binary payload to Kafka."""
         payload = self._encode_event(event)
 

mypy.ini

@@ -510,6 +510,16 @@ disallow_untyped_defs = true
 warn_return_any = true
 warn_unreachable = true
 
+[mypy-homeassistant.components.apache_kafka.*]
+check_untyped_defs = true
+disallow_incomplete_defs = true
+disallow_subclassing_any = true
+disallow_untyped_calls = true
+disallow_untyped_decorators = true
+disallow_untyped_defs = true
+warn_return_any = true
+warn_unreachable = true
+
 [mypy-homeassistant.components.apcupsd.*]
 check_untyped_defs = true
 disallow_incomplete_defs = true