Corban-Lee Jones
2b6f68de0d
All checks were successful
Build and Push Docker Image / build (push) Successful in 28s
45 lines
1.1 KiB
Python
45 lines
1.1 KiB
Python
"""
|
|
|
|
"""
|
|
|
|
import logging
|
|
from pathlib import Path
|
|
|
|
import aiofiles
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
class LogFileReader:
|
|
"""
|
|
"""
|
|
def __init__(self, file_path: Path):
|
|
if type(file_path) != Path:
|
|
raise TypeError(f"file_path must be type Path, not {type(file_path)}")
|
|
|
|
self.file_path = file_path
|
|
self._last_line_number = 0
|
|
|
|
|
|
async def read(self):
|
|
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:
|
|
if self._last_line_number == 0:
|
|
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
|
|
|
|
for line in lines:
|
|
yield line.strip()
|
|
|
|
self._last_line_number = await file.tell()
|