""" """ import logging from pathlib import Path import aiofiles log = logging.getLogger(__name__) class LogFileReader: """ """ def __init__(self, file_path: Path, track_from_start: bool=False): if not isinstance(file_path, Path): raise TypeError(f"file_path must be type Path, not {type(file_path)}") self.file_path = file_path self.track_from_start = track_from_start self._last_line_number = 0 log.debug("%s created with path %s", self.__class__.__name__, str(file_path)) async def read(self) -> list[str]: if not self.file_path.exists(): log.error("Cannot read non-existant file path: '%s'", self.file_path) raise FileNotFoundError(self.file_path) async with aiofiles.open(self.file_path, "r", encoding="utf-8") as file: log.debug("file open, and jumping to line: %s", self._last_line_number) if self._last_line_number == 0 and not self.track_from_start: await file.seek(0, 2) self._last_line_number = await file.tell() return [] await file.seek(self._last_line_number) lines = await file.readlines() if not lines: log.debug("no new lines to read") return [] self._last_line_number = await file.tell() return lines