Skip to content

Commit 3bcf6ca

Browse files
committed
Separate the FileWatcher filtering function
Separate the function to filter events so it is easier to test separately. Also remove an unnecessary `type: ignore` whole at it. Signed-off-by: Leandro Lucarella <[email protected]>
1 parent 91fbee7 commit 3bcf6ca

File tree

1 file changed

+16
-6
lines changed

1 file changed

+16
-6
lines changed

src/frequenz/channels/util/_file_watcher.py

Lines changed: 16 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -46,16 +46,26 @@ def __init__(
4646
for path in paths
4747
]
4848
self._awatch = awatch(
49-
*self._paths,
50-
stop_event=self._stop_event,
51-
watch_filter=lambda change, path_str: (
52-
change in [event_type.value for event_type in event_types] # type: ignore
53-
and pathlib.Path(path_str).is_file()
54-
),
49+
*self._paths, stop_event=self._stop_event, watch_filter=self._filter_events
5550
)
5651
self._awatch_stopped_exc: Optional[Exception] = None
5752
self._changes: Set[FileChange] = set()
5853

54+
def _filter_events(self, change: Change, path: str) -> bool:
55+
"""Filter events based on the event type and path.
56+
57+
Args:
58+
change: The type of change to be notified.
59+
path: The path of the file that changed.
60+
61+
Returns:
62+
Whether the event should be notified.
63+
"""
64+
return (
65+
change in [event_type.value for event_type in self.event_types]
66+
and pathlib.Path(path).is_file()
67+
)
68+
5969
def __del__(self) -> None:
6070
"""Cleanup registered watches.
6171

0 commit comments

Comments
 (0)