2.9.0 - Added tail, io_tail, reverse_io functions

 - Added `common.reverse_io` function, allows reading blocks of bytes from files from the end of the file efficiently.

 - Added `common.io_tail` function, a pure python generator function, which works similarly to UNIX `tail`, and
   efficiently reads the file from the end, instead of having to load the entire file to access the last lines.

 - Added `common.tail` function, which is a simple wrapper around `io_tail` - to simplify usage when tailing a relatively small ( < 10k lines? )
   amount of lines. Iterates over `io_tail`, loading each chunk into memory, and correctly orders the lines for immediate usage of the returned list.

 - Added `io_tail`, `reverse_io` and `tail` to the docs.

 - Added thorough unit tests for `io_tail` and `tail`

 - Minor cleanup of whitespace in `common.py`

Author: Someguy123

docs/source/helpers/privex.helpers.common.rst

@@ -24,14 +24,17 @@ privex.helpers.common
       env_keyval
       extract_settings
       get_function_params
+      io_tail
       is_false
       is_true
       inject_items
       parse_csv
       parse_keyval
+      random_str
+      reverse_io
       shell_quote
       stringify
-      random_str
+      tail
       _filter_params
 
 

privex/helpers/__init__.py

@@ -136,7 +136,7 @@ def _setup_logging(level=logging.WARNING):
 log = _setup_logging()
 name = 'helpers'
 
-VERSION = '2.8.1'
+VERSION = '2.9.0'
 
 
 

privex/helpers/common.py

@@ -23,6 +23,7 @@
 """
 import inspect
 import math
+import os
 import random
 import re
 import shlex
@@ -35,7 +36,7 @@
 from decimal import Decimal, getcontext
 from os import getenv as env
 from subprocess import PIPE, STDOUT
-from typing import Sequence, List, Union, Tuple, Type, Dict, Any, Iterable, Optional
+from typing import Sequence, List, Union, Tuple, Type, Dict, Any, Iterable, Optional, BinaryIO, Generator
 
 from privex.helpers import settings
 
@@ -51,8 +52,6 @@
 """All characters from a-z, A-Z, and 0-9 - for random strings where there's no risk of user font confusion"""
 
 
-
-
 def random_str(size: int = 50, chars: Sequence = SAFE_CHARS) -> str:
     """
     Generate a random string of arbitrary length using a given character set (string / list / tuple). Uses Python's 
@@ -118,9 +117,6 @@ def empty(v, zero: bool = False, itr: bool = False) -> bool:
     return False
 
 
-
-
-
 def empty_if(v: V, is_empty: K = None, not_empty: T = USE_ORIG_VAR, **kwargs) -> Union[T, K, V]:
     """
     Syntactic sugar for ``x if empty(y) else z``. If ``not_empty`` isn't specified, then the original value ``v``
@@ -757,8 +753,6 @@ def human_name(class_name: Union[str, bytes, callable, Type[object]]) -> str:
     return ''.join(new_name).strip()
 
 
-
-
 def shell_quote(*args: str) -> str:
     """
     Takes command line arguments as positional args, and properly quotes each argument to make it safe to
@@ -830,6 +824,146 @@ def call_sys(proc, *args, write: STRBYTES = None, **kwargs) -> Tuple[bytes, byte
     return stdout, stderr
 
 
+def reverse_io(f: BinaryIO, blocksize: int = 4096) -> Generator[bytes, None, None]:
+    """
+    Read file as series of blocks from end of file to start.
+
+    The data itself is in normal order, only the order of the blocks is reversed.
+    ie. "hello world" -> ["ld","wor", "lo ", "hel"]
+    Note that the file must be opened in binary mode.
+
+    Original source: https://stackoverflow.com/a/136354
+    """
+    if 'b' not in f.mode.lower():
+        raise Exception("File must be opened using binary mode.")
+    size = os.stat(f.name).st_size
+    fullblocks, lastblock = divmod(size, blocksize)
+    
+    # The first(end of file) block will be short, since this leaves
+    # the rest aligned on a blocksize boundary.  This may be more
+    # efficient than having the last (first in file) block be short
+    f.seek(-lastblock, 2)
+    yield f.read(lastblock)
+    
+    for i in range(fullblocks - 1, -1, -1):
+        f.seek(i * blocksize)
+        yield f.read(blocksize)
+
+
+def io_tail(f: BinaryIO, nlines: int = 20, bsz: int = 4096) -> Generator[List[str], None, None]:
+    """
+    NOTE: If you're only loading a small amount of lines, e.g. less than 1MB, consider using the much easier :func:`.tail`
+          function - it only requires one call and returns the lines as a singular, correctly ordered list.
+    
+    This is a generator function which works similarly to ``tail`` on UNIX systems. It efficiently retrieves lines in reverse order using
+    the passed file handle ``f``.
+    
+    WARNING: This function is a generator which returns "chunks" of lines - while the lines within each chunk are in the correct order,
+    the chunks themselves are backwards, i.e. each chunk retrieves lines prior to the previous chunk.
+    
+    This function was designed as a generator to allow for **memory efficient handling of large files**, and tailing large amounts of lines.
+    It only loads ``bsz`` bytes from the file handle into memory with each iteration, allowing you to process each chunk of lines as
+    they're read from the file, instead of having to load all ``nlines`` lines into memory at once.
+    
+    To ensure your retrieved lines are in the correct order, with each iteration you must PREPEND the outputted chunk to your final result,
+    rather than APPEND. Example::
+     
+        >>> from privex.helpers import io_tail
+        >>> lines = []
+        >>> with open('/tmp/example', 'rb') as fp:
+        ...     # We prepend each chunk from 'io_tail' to our result variable 'lines'
+        ...     for chunk in io_tail(fp, nlines=10):
+        ...         lines = chunk + lines
+        >>> print('\\n'.join(lines))
+
+    Modified to be more memory efficient, but originally based on this SO code snippet: https://stackoverflow.com/a/136354
+
+    :param BinaryIO f: An open file handle for the file to tail, must be in **binary mode** (e.g. ``rb``)
+    :param int nlines: Total number of lines to retrieve from the end of the file
+    :param int bsz:    Block size (in bytes) to load with each iteration (default: 4096 bytes). DON'T CHANGE UNLESS YOU
+                       UNDERSTAND WHAT THIS MEANS.
+    :return Generator chunks: Generates chunks (in reverse order) of correctly ordered lines as ``List[str]``
+    """
+    buf = ''
+    lines_read = 0
+    # Load 4096 bytes at a time, from file handle 'f' in reverse
+    for block in reverse_io(f, blocksize=int(bsz)):
+        # Incase we had a partial line during our previous iteration, we append leftover bytes from
+        # the previous iteration to the end of the newly loaded block
+        buf = stringify(block) + buf
+        lines = buf.splitlines()
+    
+        # Return all lines except the first (since may be partial)
+        if lines:
+            # First line may not be complete, since we're loading blocks from the bottom of the file.
+            # We yield from line 2 onwards, storing line 1 back into 'buf' to be appended to the next block.
+            result = lines[1:]
+            res_lines = len(result)
+        
+            # If we've retrieved enough lines to meet the requested 'nlines', then we just calculate how many
+            # more lines the caller wants, yield them, then return to finish execution.
+            if (lines_read + res_lines) >= nlines:
+                rem_lines = nlines - lines_read
+                lines_read += rem_lines
+                yield result[-rem_lines:]
+                return
+        
+            # Yield the lines we've loaded so far
+            if res_lines > 0:
+                lines_read += res_lines
+                yield result
+        
+            # Replace the buffer with the discarded 1st line from earlier.
+            buf = lines[0]
+    # If the loop is broken, it means we've probably reached the start of the file, and we're missing the first line...
+    # Thus we have to yield the buffer, which should contain the first line of the file.
+    yield [buf]
+
+
+def tail(filename: str, nlines: int = 20, bsz: int = 4096) -> List[str]:
+    """
+    Pure python equivalent of the UNIX ``tail`` command. Simply pass a filename and the number of lines you want to load
+    from the end of the file, and a ``List[str]`` of lines (in forward order) will be returned.
+    
+    This function is simply a wrapper for the highly efficient :func:`.io_tail`, designed for usage with a small (<10,000) amount
+    of lines to be tailed. To allow for the lines to be returned in the correct order, it must load all ``nlines`` lines into memory
+    before it can return the data.
+    
+    If you need to ``tail`` a large amount of data, e.g. 10,000+ lines of a logfile, you should consider using the lower level
+    function :func:`.io_tail` - which acts as a generator, only loading a certain amount of bytes into memory per iteration.
+    
+    Example file ``/tmp/testing``::
+        
+        this is an example 1
+        this is an example 2
+        this is an example 3
+        this is an example 4
+        this is an example 5
+        this is an example 6
+    
+    Example usage::
+    
+        >>> from privex.helpers import tail
+        >>> lines = tail('/tmp/testing', nlines=3)
+        >>> print("\\n".join(lines))
+        this is an example 4
+        this is an example 5
+        this is an example 6
+    
+    
+    :param str filename: Path to file to tail. Relative or absolute path. Absolute path is recommended for safety.
+    :param int nlines:   Total number of lines to retrieve from the end of the file
+    :param int bsz:      Block size (in bytes) to load with each iteration (default: 4096 bytes). DON'T CHANGE UNLESS YOU
+                         UNDERSTAND WHAT THIS MEANS.
+    :return List[str] lines: The last 'nlines' lines of the file 'filename' - in forward order.
+    """
+    res = []
+    with open(filename, 'rb') as fp:
+        for chunk in io_tail(f=fp, nlines=nlines, bsz=bsz):
+            res = chunk + res
+    return res
+
+
 IS_XARGS = re.compile('^\*([a-zA-Z0-9_])+$')
 """Pre-compiled regex for matching catch-all positional argument parameter names like ``*args``"""
 IS_XKWARGS = re.compile('^\*\*([a-zA-Z0-9_])+$')

tests/general/test_general.py

@@ -22,12 +22,15 @@
 
 
 """
-
+import os
 from os import path, makedirs
-from tempfile import TemporaryDirectory, NamedTemporaryFile
-from typing import Union
+from tempfile import TemporaryDirectory, NamedTemporaryFile, mkstemp
+from typing import Union, Tuple, List, TextIO, BinaryIO
 from privex import helpers
 from tests import PrivexBaseCase
+import logging
+
+log = logging.getLogger(__name__)
 
 
 class TestGeneral(PrivexBaseCase):
@@ -309,3 +312,98 @@ def test_extract_settings_case_sensitive_lowercase_keys_fail(self):
         self.assertTrue(isinstance(extracted, dict))
         self.assertEqual(len(extracted.keys()), 0)
 
+    def _create_test_file(self, tfile: BinaryIO, nlines=10) -> List[str]:
+        """Helper function for populating a testing temp file with numbered example lines for comparison"""
+        lines = [f"This is an example line {i}\n".encode('utf-8') for i in range(1, nlines+1)]
+        tfile.writelines(lines)
+        tfile.flush()
+        return [l.decode().strip("\n") for l in lines]
+
+    def test_io_tail_500_lines_300(self):
+        """
+        Test :func:`.io_tail` by tailing 300 lines of a 500 line file, then comparing each line from generated chunks against the
+        original lines written to the file.
+        """
+        with NamedTemporaryFile() as tfile:
+            lines = self._create_test_file(tfile, 500)
+            
+            i = -1  # Position -1 is the last line in the ``lines`` list
+            for chunk in helpers.io_tail(tfile, 300):
+                # We reverse each chunk, so that we can cleanly compare last lines -> first lines
+                chunk.reverse()
+                # We lower i by 1 for each line in the chunk, so we're reading ``lines`` backwards, while reading the reversed ``chunk``
+                # from the last line until the first line of the chunk.
+                for l in chunk:
+                    self.assertEqual(l, lines[i], msg=f"l == lines[{i}] // '{l}' == '{lines[i]}'")
+                    i -= 1
+            # Since the last line of ``lines`` was -1 instead of -0, the final iteration should result in -301
+            self.assertEqual(i, -301)
+
+    def test_tail_10_lines_3(self):
+        """
+        Test :func:`.tail` by comparing the last 3 lines of a 10 line testing file.
+        """
+        with NamedTemporaryFile() as tfile:
+            lines = self._create_test_file(tfile, 10)
+            
+            tailed = helpers.tail(tfile.name, 3)
+            self.assertEqual(len(tailed), 3)
+            self.assertEqual(tailed[0], "This is an example line 8")
+            self.assertEqual(tailed[1], "This is an example line 9")
+            self.assertEqual(tailed[2], "This is an example line 10")
+
+    def test_tail_10_lines_5(self):
+        """
+        Test :func:`.tail` by comparing the first and last tailed 5 lines of a 10 line testing file.
+        """
+        with NamedTemporaryFile() as tfile:
+            lines = self._create_test_file(tfile, 10)
+        
+            tailed = helpers.tail(tfile.name, 5)
+            self.assertEqual(len(tailed), 5)
+            self.assertEqual(tailed[0], lines[-5])
+            self.assertEqual(tailed[4], lines[-1])
+
+    def test_tail_10_lines_10(self):
+        """
+        Test :func:`.tail` works when ``nlines`` is equal to the amount of lines in the file. We tail 10 lines of a 10 line test file,
+        then compare all 10 original lines against the output from tail.
+        """
+        with NamedTemporaryFile() as tfile:
+            lines = self._create_test_file(tfile, 10)
+        
+            tailed = helpers.tail(tfile.name, 10)
+            self.assertEqual(len(tailed), 10)
+            for i, l in enumerate(lines):
+                self.assertEqual(tailed[i], lines[i], msg=f"tailed[{i}] == lines[{i}] // '{tailed[i]}' == '{lines[i]}'")
+
+    def test_tail_500_lines_20(self):
+        """
+        Test :func:`.tail` with a larger test file. Tailing 20 lines of a 500 line test file.
+        """
+        with NamedTemporaryFile() as tfile:
+            lines = self._create_test_file(tfile, 500)
+            
+            tailed = helpers.tail(tfile.name, 20)
+            self.assertEqual(len(tailed), 20)
+            # Compare the last 20 lines from ``lines``, against ``tailed`` starting from position 0
+            i = 0
+            for l in lines[480:]:
+                self.assertEqual(tailed[i], l, msg=f"tailed[i] == l // '{tailed[i]}' == '{l}'")
+                i += 1
+
+    def test_tail_500_lines_300(self):
+        """
+        Test :func:`.tail` with a larger line count. Tailing 300 lines of a 500 line test file.
+        """
+        with NamedTemporaryFile() as tfile:
+            lines = self._create_test_file(tfile, 500)
+        
+            tailed = helpers.tail(tfile.name, 300)
+            self.assertEqual(len(tailed), 300)
+            # Compare the last 300 lines from ``lines``, against ``tailed`` starting from position 0
+            i = 0
+            for l in lines[200:]:
+                self.assertEqual(tailed[i], l, msg=f"tailed[i] == l // '{tailed[i]}' == '{l}'")
+                i += 1
+