Improve folder_watcher typing (#107271)

homeassistant/components/folder_watcher/__init__.py

@@ -1,13 +1,25 @@
 """Component for monitoring activity on a folder."""
+from __future__ import annotations
+
 import logging
 import os
+from typing import cast
 
 import voluptuous as vol
-from watchdog.events import PatternMatchingEventHandler
+from watchdog.events import (
+    FileClosedEvent,
+    FileCreatedEvent,
+    FileDeletedEvent,
+    FileModifiedEvent,
+    FileMovedEvent,
+    FileSystemEvent,
+    FileSystemMovedEvent,
+    PatternMatchingEventHandler,
+)
 from watchdog.observers import Observer
 
 from homeassistant.const import EVENT_HOMEASSISTANT_START, EVENT_HOMEASSISTANT_STOP
-from homeassistant.core import HomeAssistant
+from homeassistant.core import Event, HomeAssistant
 import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.typing import ConfigType
 
@@ -42,8 +54,8 @@ def setup(hass: HomeAssistant, config: ConfigType) -> bool:
     """Set up the folder watcher."""
     conf = config[DOMAIN]
     for watcher in conf:
-        path = watcher[CONF_FOLDER]
+        path: str = watcher[CONF_FOLDER]
-        patterns = watcher[CONF_PATTERNS]
+        patterns: list[str] = watcher[CONF_PATTERNS]
         if not hass.config.is_allowed_path(path):
             _LOGGER.error("Folder %s is not valid or allowed", path)
             return False
@@ -52,18 +64,21 @@ def setup(hass: HomeAssistant, config: ConfigType) -> bool:
     return True
 
 
-def create_event_handler(patterns, hass):
+def create_event_handler(patterns: list[str], hass: HomeAssistant) -> EventHandler:
     """Return the Watchdog EventHandler object."""
 
-    class EventHandler(PatternMatchingEventHandler):
+    return EventHandler(patterns, hass)
+
+
+class EventHandler(PatternMatchingEventHandler):
     """Class for handling Watcher events."""
 
-        def __init__(self, patterns, hass):
+    def __init__(self, patterns: list[str], hass: HomeAssistant) -> None:
         """Initialise the EventHandler."""
         super().__init__(patterns)
         self.hass = hass
 
-        def process(self, event, moved=False):
+    def process(self, event: FileSystemEvent, moved: bool = False) -> None:
         """On Watcher event, fire HA event."""
         _LOGGER.debug("process(%s)", event)
         if not event.is_directory:
@@ -76,6 +91,7 @@ def create_event_handler(patterns, hass):
             }
 
             if moved:
+                event = cast(FileSystemMovedEvent, event)
                 dest_folder, dest_file_name = os.path.split(event.dest_path)
                 fireable.update(
                     {
@@ -89,33 +105,31 @@ def create_event_handler(patterns, hass):
                 fireable,
             )
 
-        def on_modified(self, event):
+    def on_modified(self, event: FileModifiedEvent) -> None:
         """File modified."""
         self.process(event)
 
-        def on_moved(self, event):
+    def on_moved(self, event: FileMovedEvent) -> None:
         """File moved."""
         self.process(event, moved=True)
 
-        def on_created(self, event):
+    def on_created(self, event: FileCreatedEvent) -> None:
         """File created."""
         self.process(event)
 
-        def on_deleted(self, event):
+    def on_deleted(self, event: FileDeletedEvent) -> None:
         """File deleted."""
         self.process(event)
 
-        def on_closed(self, event):
+    def on_closed(self, event: FileClosedEvent) -> None:
         """File closed."""
         self.process(event)
 
-    return EventHandler(patterns, hass)
-
 
 class Watcher:
     """Class for starting Watchdog."""
 
-    def __init__(self, path, patterns, hass):
+    def __init__(self, path: str, patterns: list[str], hass: HomeAssistant) -> None:
         """Initialise the watchdog observer."""
         self._observer = Observer()
         self._observer.schedule(
@@ -124,11 +138,11 @@ class Watcher:
         hass.bus.listen_once(EVENT_HOMEASSISTANT_START, self.startup)
         hass.bus.listen_once(EVENT_HOMEASSISTANT_STOP, self.shutdown)
 
-    def startup(self, event):
+    def startup(self, event: Event) -> None:
         """Start the watcher."""
         self._observer.start()
 
-    def shutdown(self, event):
+    def shutdown(self, event: Event) -> None:
         """Shutdown the watcher."""
         self._observer.stop()
         self._observer.join()