Created
March 12, 2024 09:12
-
-
Save mathieu-b/12107f181828926fe3c769658736cff2 to your computer and use it in GitHub Desktop.
An implementation of the 'tail -f' command in Python also supporting "file truncation" (similar to the original 'tail -f' command).
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import os | |
import sys | |
import time | |
from typing import Iterator, IO | |
def tail_follow(file_path: str, polling_interval_seconds: float = 0.1) -> Iterator[str]: | |
""" | |
Simplified 'tail -f' implementation in Python. | |
Based on https://stackoverflow.com/a/54263201 | |
Additions: | |
- handling of file permission errors | |
- handling of file truncation | |
""" | |
def _get_file_size(file_object: IO) -> int: | |
return os.fstat(file_object.fileno()).st_size | |
if not os.path.isfile(file_path): | |
print(f"File not found: '{file_path}'", file=sys.stderr) | |
sys.exit(1) | |
if not os.access(file_path, os.R_OK): | |
print(f"File not accessible: '{file_path}'", file=sys.stderr) | |
sys.exit(1) | |
try: | |
with open(file_path, "r") as file_to_follow: | |
line_buffer = "" | |
latest_size = _get_file_size(file_to_follow) | |
while True: | |
new_size = _get_file_size(file_to_follow) | |
if new_size < latest_size: | |
# As in coreutils 'tail' we seek back to zero "just in case", see: | |
# https://github.com/coreutils/coreutils/blob/master/src/tail.c#L1272 | |
print( | |
f"File truncated: '{file_to_follow.name}', resetting file pointer to file beginning.", | |
file=sys.stderr, | |
) | |
file_to_follow.seek(0) | |
latest_size = new_size | |
latest_read_line = file_to_follow.readline() | |
if latest_read_line is not None and latest_read_line != "": | |
line_buffer += latest_read_line | |
if line_buffer.endswith("\n"): | |
yield line_buffer | |
line_buffer = "" | |
elif polling_interval_seconds: | |
time.sleep(polling_interval_seconds) | |
except KeyboardInterrupt: | |
print("Interrupted by user") | |
except Exception as e: | |
print(f"An error occurred: {e}") | |
sys.exit(1) | |
if __name__ == "__main__": | |
for line in tail_follow(file_path=sys.argv[1]): | |
print(line, end="") |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment