Corban-Lee Jones
41b50e352f
All checks were successful
Build and Push Docker Image / build (push) Successful in 33s
45 lines
1.4 KiB
Python
45 lines
1.4 KiB
Python
"""
|
|
|
|
"""
|
|
|
|
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
|