Skip to content

Instantly share code, notes, and snippets.

@mathieu-b
Created March 12, 2024 09:12
Show Gist options
  • Save mathieu-b/12107f181828926fe3c769658736cff2 to your computer and use it in GitHub Desktop.
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).
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