Function observe_changes

src/quart/utils.py:105–136

(
    sleep: Callable[[float], Awaitable[Any]], shutdown_event: Event
)

async def observe_changes(
    sleep: Callable[[float], Awaitable[Any]], shutdown_event: Event
) -> None:
    last_updates: dict[Path, float] = {}
    for module in list(sys.modules.values()):
        filename = getattr(module, "__file__", None)
        if filename is None:
            continue
        path = Path(filename)
        try:
            last_updates[Path(filename)] = path.stat().st_mtime
        except (FileNotFoundError, NotADirectoryError):
            pass

    while not shutdown_event.is_set():
        await sleep(1)

        for index, (path, last_mtime) in enumerate(last_updates.items()):
            if index % 10 == 0:
                # Yield to the event loop
                await sleep(0)

            try:
                mtime = path.stat().st_mtime
            except FileNotFoundError as e:
                # File deleted
                raise MustReloadError() from e
            else:
                if mtime > last_mtime:
                    raise MustReloadError()
                else:
                    last_updates[path] = mtime


def restart() -> None: