Skip to content

file_watcher

frequenz.channels.file_watcher ¤

A receiver for watching for new, modified or deleted files.

Tip

Read the FileWatcher documentation for more information.

This module contains the following:

  • FileWatcher: A receiver that watches for file events.
  • Event: A file change event.
  • EventType: The types of file events that can be observed.

Classes¤

frequenz.channels.file_watcher.Event dataclass ¤

A file change event.

Source code in frequenz/channels/file_watcher.py
@dataclass(frozen=True)
class Event:
    """A file change event."""

    type: EventType
    """The type of change that was observed."""

    path: pathlib.Path
    """The path where the change was observed."""
Attributes¤
path instance-attribute ¤
path: Path

The path where the change was observed.

type instance-attribute ¤
type: EventType

The type of change that was observed.

frequenz.channels.file_watcher.EventType ¤

Bases: Enum

The types of file events that can be observed.

Source code in frequenz/channels/file_watcher.py
class EventType(Enum):
    """The types of file events that can be observed."""

    CREATE = Change.added
    """The file was created."""

    MODIFY = Change.modified
    """The file was modified."""

    DELETE = Change.deleted
    """The file was deleted."""
Attributes¤
CREATE class-attribute instance-attribute ¤
CREATE = added

The file was created.

DELETE class-attribute instance-attribute ¤
DELETE = deleted

The file was deleted.

MODIFY class-attribute instance-attribute ¤
MODIFY = modified

The file was modified.

frequenz.channels.file_watcher.FileWatcher ¤

Bases: Receiver[Event]

A receiver that watches for file events.

Usage¤

A FileWatcher receiver can be used to watch for changes in a set of files. It will generate an Event message every time a file is created, modified or deleted, depending on the type of events that it is configured to watch for.

The event message contains the type of change that was observed and the path where the change was observed.

Note

The owner of the FileWatcher receiver is responsible for recreating the FileWatcher after it has been cancelled or stopped. For example, if a Task uses an asynchronous iterator to consume events from the FileWatcher and the task is cancelled, the FileWatcher will also stop. Therefore, the same FileWatcher instance cannot be reused for a new task to consume events. In this case, a new FileWatcher instance must be created.

Event Types¤

The following event types are available:

  • CREATE: The file was created.
  • MODIFY: The file was modified.
  • DELETE: The file was deleted.
Example¤
Watch for changes and exit after the file is modified
import asyncio

from frequenz.channels.file_watcher import EventType, FileWatcher

PATH = "/tmp/test.txt"
file_watcher = FileWatcher(paths=[PATH], event_types=[EventType.MODIFY])


async def update_file() -> None:
    await asyncio.sleep(1)
    with open(PATH, "w", encoding="utf-8") as file:
        file.write("Hello, world!")


async def main() -> None:
    # Create file
    with open(PATH, "w", encoding="utf-8") as file:
        file.write("Hello, world!")
    async with asyncio.TaskGroup() as group:
        group.create_task(update_file())
        async for event in file_watcher:
            print(f"File {event.path}: {event.type.name}")
            break


asyncio.run(main())
Source code in frequenz/channels/file_watcher.py
class FileWatcher(Receiver[Event]):
    """A receiver that watches for file events.

    # Usage

    A [`FileWatcher`][frequenz.channels.file_watcher.FileWatcher] receiver can be used
    to watch for changes in a set of files. It will generate an
    [`Event`][frequenz.channels.file_watcher.Event] message every time a file is
    created, modified or deleted, depending on the type of events that it is configured
    to watch for.

    The [event][frequenz.channels.file_watcher.EventType] message contains the
    [`type`][frequenz.channels.file_watcher.Event.type] of change that was observed and
    the [`path`][frequenz.channels.file_watcher.Event.path] where the change was
    observed.

    Note:
        The owner of the [`FileWatcher`][frequenz.channels.file_watcher.FileWatcher]
        receiver is responsible for recreating the `FileWatcher` after it has been
        cancelled or stopped.
        For example, if a [`Task`][asyncio.Task] uses an asynchronous iterator to consume
        events from the `FileWatcher` and the task is cancelled, the `FileWatcher` will
        also stop. Therefore, the same `FileWatcher` instance cannot be reused for a new
        task to consume events. In this case, a new FileWatcher instance must be created.

    # Event Types

    The following event types are available:

    * [`CREATE`][frequenz.channels.file_watcher.EventType.CREATE]:
        {{docstring_summary("frequenz.channels.file_watcher.EventType.CREATE")}}
    * [`MODIFY`][frequenz.channels.file_watcher.EventType.MODIFY]:
        {{docstring_summary("frequenz.channels.file_watcher.EventType.MODIFY")}}
    * [`DELETE`][frequenz.channels.file_watcher.EventType.DELETE]:
        {{docstring_summary("frequenz.channels.file_watcher.EventType.DELETE")}}

    # Example

    Example: Watch for changes and exit after the file is modified
        ```python
        import asyncio

        from frequenz.channels.file_watcher import EventType, FileWatcher

        PATH = "/tmp/test.txt"
        file_watcher = FileWatcher(paths=[PATH], event_types=[EventType.MODIFY])


        async def update_file() -> None:
            await asyncio.sleep(1)
            with open(PATH, "w", encoding="utf-8") as file:
                file.write("Hello, world!")


        async def main() -> None:
            # Create file
            with open(PATH, "w", encoding="utf-8") as file:
                file.write("Hello, world!")
            async with asyncio.TaskGroup() as group:
                group.create_task(update_file())
                async for event in file_watcher:
                    print(f"File {event.path}: {event.type.name}")
                    break


        asyncio.run(main())
        ```
    """

    def __init__(
        self,
        paths: list[pathlib.Path | str],
        event_types: abc.Iterable[EventType] = frozenset(EventType),
    ) -> None:
        """Initialize this file watcher.

        Args:
            paths: The paths to watch for changes.
            event_types: The types of events to watch for. Defaults to watch for
                all event types.
        """
        self.event_types: frozenset[EventType] = frozenset(event_types)
        """The types of events to watch for."""

        self._stop_event: asyncio.Event = asyncio.Event()
        self._paths: list[pathlib.Path] = [
            path if isinstance(path, pathlib.Path) else pathlib.Path(path)
            for path in paths
        ]
        self._awatch: abc.AsyncGenerator[set[FileChange], None] = awatch(
            *self._paths, stop_event=self._stop_event, watch_filter=self._filter_events
        )
        self._awatch_stopped_exc: Exception | None = None
        self._changes: set[FileChange] = set()

    def _filter_events(
        self,
        change: Change,
        path: str,  # pylint: disable=unused-argument
    ) -> bool:
        """Filter events based on the event type and path.

        Args:
            change: The type of change to be notified.
            path: The path of the file that changed.

        Returns:
            Whether the event should be notified.
        """
        return change in [event_type.value for event_type in self.event_types]

    def __del__(self) -> None:
        """Finalize this file watcher."""
        # We need to set the stop event to make sure that the awatch background task
        # is stopped.
        self._stop_event.set()

    async def ready(self) -> bool:
        """Wait until the receiver is ready with a message or an error.

        Once a call to `ready()` has finished, the message should be read with
        a call to `consume()` (`receive()` or iterated over). The receiver will
        remain ready (this method will return immediately) until it is
        consumed.

        Returns:
            Whether the receiver is still active.
        """
        # if there are messages waiting to be consumed, return immediately.
        if self._changes:
            return True

        # if it was already stopped, return immediately.
        if self._awatch_stopped_exc is not None:
            return False

        try:
            self._changes = await anext(self._awatch)
        except StopAsyncIteration as err:
            self._awatch_stopped_exc = err
            return False

        return True

    def consume(self) -> Event:
        """Return the latest event once `ready` is complete.

        Returns:
            The next event that was received.

        Raises:
            ReceiverStoppedError: If there is some problem with the receiver.
        """
        if not self._changes and self._awatch_stopped_exc is not None:
            raise ReceiverStoppedError(self) from self._awatch_stopped_exc

        assert self._changes, "`consume()` must be preceded by a call to `ready()`"
        # Tuple of (Change, path) returned by watchfiles
        change, path_str = self._changes.pop()
        return Event(type=EventType(change), path=pathlib.Path(path_str))

    def __str__(self) -> str:
        """Return a string representation of this receiver."""
        if len(self._paths) > 3:
            paths = [str(p) for p in self._paths[:3]]
            paths.append("…")
        else:
            paths = [str(p) for p in self._paths]
        event_types = [event_type.name for event_type in self.event_types]
        return f"{type(self).__name__}:{','.join(event_types)}:{','.join(paths)}"

    def __repr__(self) -> str:
        """Return a string representation of this receiver."""
        return f"{type(self).__name__}({self._paths!r}, {self.event_types!r})"
Attributes¤
event_types instance-attribute ¤

The types of events to watch for.

Functions¤
__aiter__ ¤
__aiter__() -> Self

Get an async iterator over the received messages.

RETURNS DESCRIPTION
Self

This receiver, as it is already an async iterator.

Source code in frequenz/channels/_receiver.py
def __aiter__(self) -> Self:
    """Get an async iterator over the received messages.

    Returns:
        This receiver, as it is already an async iterator.
    """
    return self
__anext__ async ¤
__anext__() -> ReceiverMessageT_co

Await the next message in the async iteration over received messages.

RETURNS DESCRIPTION
ReceiverMessageT_co

The next received message.

RAISES DESCRIPTION
StopAsyncIteration

If the receiver stopped producing messages.

ReceiverError

If there is some problem with the receiver.

Source code in frequenz/channels/_receiver.py
async def __anext__(self) -> ReceiverMessageT_co:
    """Await the next message in the async iteration over received messages.

    Returns:
        The next received message.

    Raises:
        StopAsyncIteration: If the receiver stopped producing messages.
        ReceiverError: If there is some problem with the receiver.
    """
    try:
        await self.ready()
        return self.consume()
    except ReceiverStoppedError as exc:
        raise StopAsyncIteration() from exc
__del__ ¤
__del__() -> None

Finalize this file watcher.

Source code in frequenz/channels/file_watcher.py
def __del__(self) -> None:
    """Finalize this file watcher."""
    # We need to set the stop event to make sure that the awatch background task
    # is stopped.
    self._stop_event.set()
__init__ ¤
__init__(
    paths: list[Path | str],
    event_types: Iterable[EventType] = frozenset(EventType),
) -> None

Initialize this file watcher.

PARAMETER DESCRIPTION
paths

The paths to watch for changes.

TYPE: list[Path | str]

event_types

The types of events to watch for. Defaults to watch for all event types.

TYPE: Iterable[EventType] DEFAULT: frozenset(EventType)

Source code in frequenz/channels/file_watcher.py
def __init__(
    self,
    paths: list[pathlib.Path | str],
    event_types: abc.Iterable[EventType] = frozenset(EventType),
) -> None:
    """Initialize this file watcher.

    Args:
        paths: The paths to watch for changes.
        event_types: The types of events to watch for. Defaults to watch for
            all event types.
    """
    self.event_types: frozenset[EventType] = frozenset(event_types)
    """The types of events to watch for."""

    self._stop_event: asyncio.Event = asyncio.Event()
    self._paths: list[pathlib.Path] = [
        path if isinstance(path, pathlib.Path) else pathlib.Path(path)
        for path in paths
    ]
    self._awatch: abc.AsyncGenerator[set[FileChange], None] = awatch(
        *self._paths, stop_event=self._stop_event, watch_filter=self._filter_events
    )
    self._awatch_stopped_exc: Exception | None = None
    self._changes: set[FileChange] = set()
__repr__ ¤
__repr__() -> str

Return a string representation of this receiver.

Source code in frequenz/channels/file_watcher.py
def __repr__(self) -> str:
    """Return a string representation of this receiver."""
    return f"{type(self).__name__}({self._paths!r}, {self.event_types!r})"
__str__ ¤
__str__() -> str

Return a string representation of this receiver.

Source code in frequenz/channels/file_watcher.py
def __str__(self) -> str:
    """Return a string representation of this receiver."""
    if len(self._paths) > 3:
        paths = [str(p) for p in self._paths[:3]]
        paths.append("…")
    else:
        paths = [str(p) for p in self._paths]
    event_types = [event_type.name for event_type in self.event_types]
    return f"{type(self).__name__}:{','.join(event_types)}:{','.join(paths)}"
consume ¤
consume() -> Event

Return the latest event once ready is complete.

RETURNS DESCRIPTION
Event

The next event that was received.

RAISES DESCRIPTION
ReceiverStoppedError

If there is some problem with the receiver.

Source code in frequenz/channels/file_watcher.py
def consume(self) -> Event:
    """Return the latest event once `ready` is complete.

    Returns:
        The next event that was received.

    Raises:
        ReceiverStoppedError: If there is some problem with the receiver.
    """
    if not self._changes and self._awatch_stopped_exc is not None:
        raise ReceiverStoppedError(self) from self._awatch_stopped_exc

    assert self._changes, "`consume()` must be preceded by a call to `ready()`"
    # Tuple of (Change, path) returned by watchfiles
    change, path_str = self._changes.pop()
    return Event(type=EventType(change), path=pathlib.Path(path_str))
filter ¤
filter(
    filter_function: Callable[[ReceiverMessageT_co], bool]
) -> Receiver[ReceiverMessageT_co]

Apply a filter function on the messages on a receiver.

Tip

The returned receiver type won't have all the methods of the original receiver. If you need to access methods of the original receiver that are not part of the Receiver interface you should save a reference to the original receiver and use that instead.

PARAMETER DESCRIPTION
filter_function

The function to be applied on incoming messages to determine if they should be received.

TYPE: Callable[[ReceiverMessageT_co], bool]

RETURNS DESCRIPTION
Receiver[ReceiverMessageT_co]

A new receiver that only receives messages that pass the filter.

Source code in frequenz/channels/_receiver.py
def filter(
    self, filter_function: Callable[[ReceiverMessageT_co], bool], /
) -> Receiver[ReceiverMessageT_co]:
    """Apply a filter function on the messages on a receiver.

    Tip:
        The returned receiver type won't have all the methods of the original
        receiver. If you need to access methods of the original receiver that are
        not part of the `Receiver` interface you should save a reference to the
        original receiver and use that instead.

    Args:
        filter_function: The function to be applied on incoming messages to
            determine if they should be received.

    Returns:
        A new receiver that only receives messages that pass the filter.
    """
    return _Filter(receiver=self, filter_function=filter_function)
map ¤

Apply a mapping function on the received message.

Tip

The returned receiver type won't have all the methods of the original receiver. If you need to access methods of the original receiver that are not part of the Receiver interface you should save a reference to the original receiver and use that instead.

PARAMETER DESCRIPTION
mapping_function

The function to be applied on incoming messages.

TYPE: Callable[[ReceiverMessageT_co], MappedMessageT_co]

RETURNS DESCRIPTION
Receiver[MappedMessageT_co]

A new receiver that applies the function on the received messages.

Source code in frequenz/channels/_receiver.py
def map(
    self, mapping_function: Callable[[ReceiverMessageT_co], MappedMessageT_co], /
) -> Receiver[MappedMessageT_co]:
    """Apply a mapping function on the received message.

    Tip:
        The returned receiver type won't have all the methods of the original
        receiver. If you need to access methods of the original receiver that are
        not part of the `Receiver` interface you should save a reference to the
        original receiver and use that instead.

    Args:
        mapping_function: The function to be applied on incoming messages.

    Returns:
        A new receiver that applies the function on the received messages.
    """
    return _Mapper(receiver=self, mapping_function=mapping_function)
ready async ¤
ready() -> bool

Wait until the receiver is ready with a message or an error.

Once a call to ready() has finished, the message should be read with a call to consume() (receive() or iterated over). The receiver will remain ready (this method will return immediately) until it is consumed.

RETURNS DESCRIPTION
bool

Whether the receiver is still active.

Source code in frequenz/channels/file_watcher.py
async def ready(self) -> bool:
    """Wait until the receiver is ready with a message or an error.

    Once a call to `ready()` has finished, the message should be read with
    a call to `consume()` (`receive()` or iterated over). The receiver will
    remain ready (this method will return immediately) until it is
    consumed.

    Returns:
        Whether the receiver is still active.
    """
    # if there are messages waiting to be consumed, return immediately.
    if self._changes:
        return True

    # if it was already stopped, return immediately.
    if self._awatch_stopped_exc is not None:
        return False

    try:
        self._changes = await anext(self._awatch)
    except StopAsyncIteration as err:
        self._awatch_stopped_exc = err
        return False

    return True
receive async ¤
receive() -> ReceiverMessageT_co

Receive a message.

RETURNS DESCRIPTION
ReceiverMessageT_co

The received message.

RAISES DESCRIPTION
ReceiverStoppedError

If there is some problem with the receiver.

ReceiverError

If there is some problem with the receiver.

Source code in frequenz/channels/_receiver.py
async def receive(self) -> ReceiverMessageT_co:
    """Receive a message.

    Returns:
        The received message.

    Raises:
        ReceiverStoppedError: If there is some problem with the receiver.
        ReceiverError: If there is some problem with the receiver.
    """
    try:
        received = await self.__anext__()  # pylint: disable=unnecessary-dunder-call
    except StopAsyncIteration as exc:
        # If we already had a cause and it was the receiver was stopped,
        # then reuse that error, as StopAsyncIteration is just an artifact
        # introduced by __anext__.
        if (
            isinstance(exc.__cause__, ReceiverStoppedError)
            # pylint is not smart enough to figure out we checked above
            # this is a ReceiverStoppedError and thus it does have
            # a receiver member
            and exc.__cause__.receiver is self  # pylint: disable=no-member
        ):
            raise exc.__cause__
        raise ReceiverStoppedError(self) from exc
    return received