[events] Use dataclass (gorakhargosh#987)

* [events] Use `dataclass`

- [events] `FileSystemEvent`, and subclasses, are now `dataclass`es, and their `repr()` has changed
- [windows] `WinAPINativeEvent` is now a `dataclass`, and its `repr()` has changed

* chore: run black with line-lentgh=120

changelog.rst

@@ -8,12 +8,13 @@ Changelog
 
 2023-xx-xx • `full history <https://github.com/gorakhargosh/watchdog/compare/v3.0.0...HEAD>`__
 
+- [events] ``FileSystemEvent``, and subclasses, are now ``dataclass``es, and their ``repr()`` has changed
+- [windows] ``WinAPINativeEvent`` is now a ``dataclass``, and its ``repr()`` has changed
 - [events] Log ``FileOpenedEvent``, and ``FileClosedEvent``, events in ``LoggingEventHandler``
 - [tests] Improve ``FileSystemEvent`` coverage
 - [watchmedo] Log all events in ``LoggerTrick``
 - Thanks to our beloved contributors: @BoboTiG
 
-
 3.0.0
 ~~~~~
 

src/watchdog/events.py

@@ -96,6 +96,7 @@
 import logging
 import os.path
 import re
+from dataclasses import dataclass, field
 from typing import Optional
 
 from watchdog.utils.patterns import match_any_paths
@@ -108,6 +109,7 @@
 EVENT_TYPE_OPENED = "opened"
 
 
+@dataclass(unsafe_hash=True)
 class FileSystemEvent:
     """
     Immutable type that represents a file system event that is triggered
@@ -117,76 +119,24 @@ class FileSystemEvent:
     can be used as keys in dictionaries or be added to sets.
     """
 
-    event_type = ""
-    """The type of the event as a string."""
+    src_path: str
+    dest_path: str = ""
+    event_type: str = field(default="", init=False)
+    is_directory: bool = field(default=False, init=False)
 
-    is_directory = False
-    """True if event was emitted for a directory; False otherwise."""
-
-    is_synthetic = False
     """
     True if event was synthesized; False otherwise.
-
     These are events that weren't actually broadcast by the OS, but
     are presumed to have happened based on other, actual events.
     """
-
-    def __init__(self, src_path):
-        self._src_path = src_path
-
-    @property
-    def src_path(self):
-        """Source path of the file system object that triggered this event."""
-        return self._src_path
-
-    def __str__(self):
-        return self.__repr__()
-
-    def __repr__(self):
-        return (
-            f"<{type(self).__name__}: event_type={self.event_type}, "
-            f"src_path={self.src_path!r}, is_directory={self.is_directory}>"
-        )
-
-    # Used for comparison of events.
-    @property
-    def key(self):
-        return (self.event_type, self.src_path, self.is_directory)
-
-    def __eq__(self, event):
-        return self.key == event.key
-
-    def __hash__(self):
-        return hash(self.key)
+    is_synthetic: bool = field(default=False)
 
 
 class FileSystemMovedEvent(FileSystemEvent):
-    """
-    File system event representing any kind of file system movement.
-    """
+    """File system event representing any kind of file system movement."""
 
     event_type = EVENT_TYPE_MOVED
 
-    def __init__(self, src_path, dest_path):
-        super().__init__(src_path)
-        self._dest_path = dest_path
-
-    @property
-    def dest_path(self):
-        """The destination path of the move event."""
-        return self._dest_path
-
-    # Used for hashing this as an immutable object.
-    @property
-    def key(self):
-        return (self.event_type, self.src_path, self.dest_path, self.is_directory)
-
-    def __repr__(self):
-        return (
-            f"<{type(self).__name__}: src_path={self.src_path!r}, "
-            f"dest_path={self.dest_path!r}, is_directory={self.is_directory}>"
-        )
-
 
 # File events.
 
@@ -519,9 +469,7 @@ def on_moved(self, event: FileSystemEvent) -> None:
         super().on_moved(event)
 
         what = "directory" if event.is_directory else "file"
-        self.logger.info(
-            "Moved %s: from %s to %s", what, event.src_path, event.dest_path  # type: ignore[attr-defined]
-        )
+        self.logger.info("Moved %s: from %s to %s", what, event.src_path, event.dest_path)
 
     def on_created(self, event: FileSystemEvent) -> None:
         super().on_created(event)
@@ -568,20 +516,12 @@ def generate_sub_moved_events(src_dir_path, dest_dir_path):
     for root, directories, filenames in os.walk(dest_dir_path):
         for directory in directories:
             full_path = os.path.join(root, directory)
-            renamed_path = (
-                full_path.replace(dest_dir_path, src_dir_path) if src_dir_path else None
-            )
-            dir_moved_event = DirMovedEvent(renamed_path, full_path)
-            dir_moved_event.is_synthetic = True
-            yield dir_moved_event
+            renamed_path = full_path.replace(dest_dir_path, src_dir_path) if src_dir_path else ""
+            yield DirMovedEvent(renamed_path, full_path, is_synthetic=True)
         for filename in filenames:
             full_path = os.path.join(root, filename)
-            renamed_path = (
-                full_path.replace(dest_dir_path, src_dir_path) if src_dir_path else None
-            )
-            file_moved_event = FileMovedEvent(renamed_path, full_path)
-            file_moved_event.is_synthetic = True
-            yield file_moved_event
+            renamed_path = full_path.replace(dest_dir_path, src_dir_path) if src_dir_path else ""
+            yield FileMovedEvent(renamed_path, full_path, is_synthetic=True)
 
 
 def generate_sub_created_events(src_dir_path):
@@ -597,10 +537,6 @@ def generate_sub_created_events(src_dir_path):
     """
     for root, directories, filenames in os.walk(src_dir_path):
         for directory in directories:
-            dir_created_event = DirCreatedEvent(os.path.join(root, directory))
-            dir_created_event.is_synthetic = True
-            yield dir_created_event
+            yield DirCreatedEvent(os.path.join(root, directory), is_synthetic=True)
         for filename in filenames:
-            file_created_event = FileCreatedEvent(os.path.join(root, filename))
-            file_created_event.is_synthetic = True
-            yield file_created_event
+            yield FileCreatedEvent(os.path.join(root, filename), is_synthetic=True)

src/watchdog/observers/__init__.py

@@ -74,6 +74,7 @@
     except Exception:
         try:
             from .kqueue import KqueueObserver as Observer
+
             warnings.warn("Failed to import fsevents. Fall back to kqueue")
         except Exception:
             from .polling import PollingObserver as Observer

src/watchdog/observers/api.py

@@ -294,9 +294,7 @@ def schedule(self, event_handler, path, recursive=False):
 
             # If we don't have an emitter for this watch already, create it.
             if self._emitter_for_watch.get(watch) is None:
-                emitter = self._emitter_class(
-                    event_queue=self.event_queue, watch=watch, timeout=self.timeout
-                )
+                emitter = self._emitter_class(event_queue=self.event_queue, watch=watch, timeout=self.timeout)
                 if self.is_alive():
                     emitter.start()
                 self._add_emitter(emitter)

src/watchdog/observers/fsevents.py

@@ -85,9 +85,7 @@ def __init__(
         self._start_time = 0.0
         self._starting_state = None
         self._lock = threading.Lock()
-        self._absolute_watch_path = os.path.realpath(
-            os.path.abspath(os.path.expanduser(self.watch.path))
-        )
+        self._absolute_watch_path = os.path.realpath(os.path.abspath(os.path.expanduser(self.watch.path)))
 
     def on_thread_stop(self):
         _fsevents.remove_watch(self.watch)
@@ -107,9 +105,7 @@ def queue_event(self, event):
                 logger.debug("drop event %s", event)
 
     def _is_recursive_event(self, event):
-        src_path = (
-            event.src_path if event.is_directory else os.path.dirname(event.src_path)
-        )
+        src_path = event.src_path if event.is_directory else os.path.dirname(event.src_path)
         if src_path == self._absolute_watch_path:
             return False
 
@@ -136,9 +132,7 @@ def _queue_modified_event(self, event, src_path, dirname):
         cls = DirModifiedEvent if event.is_directory else FileModifiedEvent
         self.queue_event(cls(src_path))
 
-    def _queue_renamed_event(
-        self, src_event, src_path, dst_path, src_dirname, dst_dirname
-    ):
+    def _queue_renamed_event(self, src_event, src_path, dst_path, src_dirname, dst_dirname):
         cls = DirMovedEvent if src_event.is_directory else FileMovedEvent
         dst_path = self._encode_path(dst_path)
         self.queue_event(cls(src_path, dst_path))
@@ -170,9 +164,7 @@ def _is_meta_mod(event):
     def queue_events(self, timeout, events):
         if logger.getEffectiveLevel() <= logging.DEBUG:
             for event in events:
-                flags = ", ".join(
-                    attr for attr in dir(event) if getattr(event, attr) is True
-                )
+                flags = ", ".join(attr for attr in dir(event) if getattr(event, attr) is True)
                 logger.debug(f"{event}: {flags}")
 
         if time.monotonic() - self._start_time > 60:
@@ -238,9 +230,7 @@ def queue_events(self, timeout, events):
                 if event.is_renamed:
                     # Check if we have a corresponding destination event in the watched path.
                     dst_event = next(
-                        iter(
-                            e for e in events if e.is_renamed and e.inode == event.inode
-                        ),
+                        iter(e for e in events if e.is_renamed and e.inode == event.inode),
                         None,
                     )
 
@@ -251,9 +241,7 @@ def queue_events(self, timeout, events):
                         dst_path = self._encode_path(dst_event.path)
                         dst_dirname = os.path.dirname(dst_path)
 
-                        self._queue_renamed_event(
-                            event, src_path, dst_path, src_dirname, dst_dirname
-                        )
+                        self._queue_renamed_event(event, src_path, dst_path, src_dirname, dst_dirname)
                         self._fs_view.add(event.inode)
 
                         for sub_event in generate_sub_moved_events(src_path, dst_path):

src/watchdog/observers/fsevents2.py

@@ -106,9 +106,7 @@ def __init__(self, path):
     def run(self):
         pool = AppKit.NSAutoreleasePool.alloc().init()
         self._run_loop = CFRunLoopGetCurrent()
-        FSEventStreamScheduleWithRunLoop(
-            self._stream_ref, self._run_loop, kCFRunLoopDefaultMode
-        )
+        FSEventStreamScheduleWithRunLoop(self._stream_ref, self._run_loop, kCFRunLoopDefaultMode)
         if not FSEventStreamStart(self._stream_ref):
             FSEventStreamInvalidate(self._stream_ref)
             FSEventStreamRelease(self._stream_ref)
@@ -126,13 +124,8 @@ def stop(self):
         if self._run_loop is not None:
             CFRunLoopStop(self._run_loop)
 
-    def _callback(
-        self, streamRef, clientCallBackInfo, numEvents, eventPaths, eventFlags, eventIDs
-    ):
-        events = [
-            NativeEvent(path, flags, _id)
-            for path, flags, _id in zip(eventPaths, eventFlags, eventIDs)
-        ]
+    def _callback(self, streamRef, clientCallBackInfo, numEvents, eventPaths, eventFlags, eventIDs):
+        events = [NativeEvent(path, flags, _id) for path, flags, _id in zip(eventPaths, eventFlags, eventIDs)]
         logger.debug(f"FSEvents callback. Got {numEvents} events:")
         for e in events:
             logger.debug(e)
@@ -219,17 +212,11 @@ def queue_events(self, timeout):
                 # from a single move operation. (None of this is documented!)
                 # Otherwise, guess whether file was moved in or out.
                 # TODO: handle id wrapping
-                if (
-                    i + 1 < len(events)
-                    and events[i + 1].is_renamed
-                    and events[i + 1].event_id == event.event_id + 1
-                ):
+                if i + 1 < len(events) and events[i + 1].is_renamed and events[i + 1].event_id == event.event_id + 1:
                     cls = DirMovedEvent if event.is_directory else FileMovedEvent
                     self.queue_event(cls(event.path, events[i + 1].path))
                     self.queue_event(DirModifiedEvent(os.path.dirname(event.path)))
-                    self.queue_event(
-                        DirModifiedEvent(os.path.dirname(events[i + 1].path))
-                    )
+                    self.queue_event(DirModifiedEvent(os.path.dirname(events[i + 1].path)))
                     i += 1
                 elif os.path.exists(event.path):
                     cls = DirCreatedEvent if event.is_directory else FileCreatedEvent

src/watchdog/observers/inotify.py

@@ -155,7 +155,7 @@ def queue_events(self, timeout, full_events=False):
             if event.is_moved_to:
                 if full_events:
                     cls = DirMovedEvent if event.is_directory else FileMovedEvent
-                    self.queue_event(cls(None, src_path))
+                    self.queue_event(cls("", src_path))
                 else:
                     cls = DirCreatedEvent if event.is_directory else FileCreatedEvent
                     self.queue_event(cls(src_path))
@@ -175,7 +175,7 @@ def queue_events(self, timeout, full_events=False):
                 self.queue_event(DirModifiedEvent(os.path.dirname(src_path)))
             elif event.is_moved_from and full_events:
                 cls = DirMovedEvent if event.is_directory else FileMovedEvent
-                self.queue_event(cls(src_path, None))
+                self.queue_event(cls(src_path, ""))
                 self.queue_event(DirModifiedEvent(os.path.dirname(src_path)))
             elif event.is_create:
                 cls = DirCreatedEvent if event.is_directory else FileCreatedEvent

src/watchdog/observers/inotify_buffer.py

@@ -59,11 +59,7 @@ def _group_events(self, event_list):
             logger.debug("in-event %s", inotify_event)
 
             def matching_from_event(event):
-                return (
-                    not isinstance(event, tuple)
-                    and event.is_moved_from
-                    and event.cookie == inotify_event.cookie
-                )
+                return not isinstance(event, tuple) and event.is_moved_from and event.cookie == inotify_event.cookie
 
             if inotify_event.is_moved_to:
                 # Check if move_from is already in the buffer
@@ -104,9 +100,7 @@ def run(self):
                     continue
 
                 # Only add delay for unmatched move_from events
-                delay = (
-                    not isinstance(inotify_event, tuple) and inotify_event.is_moved_from
-                )
+                delay = not isinstance(inotify_event, tuple) and inotify_event.is_moved_from
                 self._queue.put(inotify_event, delay)
 
                 if (