utils.py 883 B

12345678910111213141516171819202122232425262728293031
  1. import logging
  2. import os
  3. from typing import Iterator, Optional
  4. logger = logging.getLogger(__name__)
  5. def file_tail_iterator(path: str) -> Iterator[Optional[str]]:
  6. """Yield lines from a file as it's written.
  7. Returns lines in batches opportunistically.
  8. Returns None until the file exists or if no new line has been written.
  9. """
  10. if not isinstance(path, str):
  11. raise TypeError(f"path must be a string, got {type(path)}.")
  12. while not os.path.exists(path):
  13. logger.debug(f"Path {path} doesn't exist yet.")
  14. yield None
  15. with open(path, "r") as f:
  16. lines = ""
  17. while True:
  18. curr_line = f.readline()
  19. # readline() returns empty string when there's no new line.
  20. if curr_line:
  21. lines += curr_line
  22. else:
  23. yield lines or None
  24. lines = ""