import asyncio import os from pathlib import Path from typing import Callable, Awaitable, Optional class LogWatcher: POLL_INTERVAL = 0.1 FILE_RETRY_INTERVAL = 1.0 def __init__( self, path: str, on_line: Callable[[str], Awaitable[None]], on_offline: Optional[Callable[[], Awaitable[None]]] = None, offline_timeout: float = 60.0, ): self._path = Path(path) self._on_line = on_line self._on_offline = on_offline self._offline_timeout = offline_timeout self._task: Optional[asyncio.Task] = None def start(self) -> asyncio.Task: self._task = asyncio.create_task(self._run()) return self._task def stop(self): if self._task: self._task.cancel() async def _run(self): offline_since: Optional[float] = None while True: try: with open(self._path, 'r', errors='replace') as f: f.seek(0, 2) # start at end of file (tail -f behaviour) offline_since = None while True: line = f.readline() if line: await self._on_line(line) else: await asyncio.sleep(self.POLL_INTERVAL) except OSError: now = asyncio.get_event_loop().time() if offline_since is None: offline_since = now elif now - offline_since >= self._offline_timeout and self._on_offline: await self._on_offline() offline_since = None # reset so we don't spam await asyncio.sleep(min(self.FILE_RETRY_INTERVAL, self._offline_timeout)) except asyncio.CancelledError: return