Bobobk

Practical Python, Linux and automation guides

Read the Last N Log Lines in Python with deque

2026-10-27 Technology
Read the Last N Log Lines in Python with deque
On this page

A log is large, but a failure report only needs its last 20 lines. Using readlines()[-20:] first loads the entire file into memory. This tutorial uses the standard-library deque to retain the last N lines and verifies Unicode, blank lines, and a final line without a newline.

The examples target beginner-to-intermediate readers and completed, ordinary UTF-8 log files that stay unchanged during reading. For a growing active log, FIFO, or device file, obtain a stable regular-file copy first.

1. Method: read sequentially and retain the final window

Pass the file iterator to deque(source, maxlen=N). Once full, the queue discards the oldest lines. The official Python deque recipes demonstrate this approach.

This still scans the whole file from the beginning, so runtime grows with input length. The queue retains at most N lines, but a line can be huge: the limit counts lines rather than bytes, and reading an unusually long line also needs corresponding memory. For a quick terminal inspection of a large regular file, use the system command tail -n 20 task.log; the implementation below is useful inside Python automation.

2. Setup: create a repeatable sample log

Use Python 3.10 or newer. No third-party packages are required. Commands were checked on Linux with Python 3.11.4; shell examples work in bash and zsh.

  1. Create a five-line log in a new directory. Deliberately omit the final newline:
python3 --version
mkdir tail-demo
cd tail-demo
python3 - <<'PYCODE'
from pathlib import Path
Path("task.log").write_text(
    "01 INFO start\n02 INFO loaded\n03 WARNING retry\n"
    "04 INFO recovered\n05 ERROR 连接失败",
    encoding="utf-8",
)
PYCODE
  1. Save the next section as tail_log.py in the current directory. Run subsequent examples from this demo directory.

3. Implementation: a complete command-line tool

"""Print the last N lines of a completed UTF-8 log. Python 3.10+."""
import argparse
from collections import deque
from pathlib import Path
import sys


def positive_int(value):
    number = int(value)
    if number < 1:
        raise argparse.ArgumentTypeError("must be at least 1")
    return number


def tail_lines(path, count):
    if count < 1:
        raise ValueError("count must be at least 1")
    with Path(path).open(encoding="utf-8", errors="strict") as source:
        return deque(source, maxlen=count)


def main():
    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
    parser.add_argument("path", type=Path, help="completed UTF-8 text log")
    parser.add_argument("-n", "--lines", type=positive_int, default=20)
    args = parser.parse_args()
    try:
        recent = tail_lines(args.path, args.lines)
        sys.stdout.writelines(recent)
        sys.stdout.flush()
    except (OSError, UnicodeError) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

The script finishes reading before emitting output. A bad input path or invalid UTF-8 anywhere in the file returns 1 with empty stdout, including invalid bytes outside the retained window. Invalid CLI arguments return 2; success returns 0. An empty file succeeds with no output.

Reading uses explicit, strict UTF-8 decoding. The Python open documentation describes default text newline handling: CRLF and bare CR become \n. Output therefore preserves text content and whether the final line has a terminator, rather than preserving all original newline bytes. It adds no timestamps or line numbers.

writelines() inserts no extra separators. An output write failure can still leave partial output, so automation must check the exit status. Choose a fresh report path for redirection. Never redirect to the input log itself: the shell would truncate it before Python reads it.

4. Usage: select the last three lines, then filter errors

  1. Inspect help, select three lines, and check the output bytes:
python3 tail_log.py --help
python3 -X utf8 tail_log.py -n 3 task.log > recent.log
python3 - <<'PYCODE'
from pathlib import Path
expected = "03 WARNING retry\n04 INFO recovered\n05 ERROR 连接失败"
assert Path("recent.log").read_bytes() == expected.encode("utf-8")
print("PASS: last 3 lines, UTF-8, no added newline")
PYCODE

Expect PASS: last 3 lines, UTF-8, no added newline. -X utf8 selects UTF-8 for this invocation’s standard output; the file-reading encoding remains explicit in the code. If only Unicode output fails in a scheduled task, follow the standard-output encoding troubleshooting guide.

  1. Import the function to build a Python report. Save this as recent_errors.py and run python3 -X utf8 recent_errors.py:
from tail_log import tail_lines

for line in tail_lines("task.log", 3):
    if "ERROR" in line:
        print(line, end="")

Output is 05 ERROR 连接失败, still without a final newline. This selects the last three lines before filtering ERROR. Filtering the whole input first would instead select the last three errors, which answers a different question. When imported, the function propagates read errors for the calling job to handle.

5. Validation: boundary inputs and failure statuses

Save this as check_tail.py and run python3 check_tail.py. Fixtures live in a temporary directory that is cleaned automatically:

from pathlib import Path
import subprocess
import sys
from tempfile import TemporaryDirectory

script = Path("tail_log.py").resolve()


def run(path, count):
    return subprocess.run(
        [sys.executable, "-X", "utf8", str(script), "-n", str(count), str(path)],
        capture_output=True,
        timeout=10,
    )


with TemporaryDirectory(prefix="tail-check-") as workspace:
    folder = Path(workspace)
    cases = [
        (b"", 3, b""),
        (b"one\n", 5, b"one\n"),
        (b"a\nb\nc", 2, b"b\nc"),
        (b"a\n\nb\n", 2, b"\nb\n"),
        ("甲\n乙\n丙".encode("utf-8"), 1, "丙".encode("utf-8")),
        (b"a\r\nb\r\n", 1, b"b\n"),
    ]
    for index, (content, count, expected) in enumerate(cases):
        path = folder / f"case {index}.log"
        path.write_bytes(content)
        result = run(path, count)
        assert (result.returncode, result.stdout, result.stderr) == (0, expected, b"")
        assert path.read_bytes() == content

    bad = folder / "invalid.log"
    bad.write_bytes(b"bad:\xff\nlater valid line\n")
    for path in (bad, folder / "missing.log", folder):
        result = run(path, 1)
        assert result.returncode == 1 and result.stdout == b""
        assert b"ERROR:" in result.stderr

    for count in (0, -1, "oops"):
        result = run(folder / "case 0.log", count)
        assert result.returncode == 2 and result.stdout == b""

print("PASS: content, line endings, input preserved, read errors, bad arguments")

Expect PASS: content, line endings, input preserved, read errors, bad arguments. Coverage includes paths containing spaces, empty and short files, blank lines, Unicode, CRLF, and invalid encoding before the retained window.

Then run this slightly larger example to check the retained lines:

python3 - <<'PYCODE'
from pathlib import Path
from tail_log import tail_lines

path = Path("many.log")
with path.open("x", encoding="utf-8") as output:
    for index in range(100_000):
        output.write(f"row={index}\n")
recent = tail_lines(path, 2)
assert list(recent) == ["row=99998\n", "row=99999\n"]
assert recent.maxlen == 2
print("PASS: retained 2 of 100000 lines")
PYCODE

Expect PASS: retained 2 of 100000 lines. Exclusive x mode prevents overwriting many.log on a repeated run; use a fresh directory to repeat it. This checks retained content and line count, rather than measuring memory performance.

6. Troubleshooting and scope

Problem Action
No such file or directory Check the working directory and pass an absolute log path in automation.
Permission denied Check file and parent-directory permissions as the job’s account; select a log that account can read.
UTF-8 decoding failure Confirm the producer’s encoding, change the explicit reader encoding, and retest. Avoid silently dropping text with errors="ignore".
Unicode output encoding error Use the demonstrated python3 -X utf8 invocation and inspect that process’s standard streams.
Reading is slow This algorithm scans the full input. Reduce the input scope, or use system tail for simple end-of-file inspection.
A business event is cut in half The tool counts text lines. Multiline tracebacks can be split; parse business records if whole events are required.

The tool neither follows new writes nor combines rotated backups. Rewrites or rotation during reading do not provide a consistent snapshot. To control the producer’s log size, configure size-based log rotation. For .gz backups, use the streaming gzip log and integrity-check tutorial rather than opening compressed bytes as ordinary text.

7. Summary

A deque retains the last N lines without storing the entire log, making it useful for failure summaries. Verify line endings, encoding, and error statuses before adding the helper to automated reports. Account for both full-file scanning time and the memory required by unusually long lines.

Tags

1024 2fa ablang adsense agents.md ai ai-agent ai-agents ai-seo algorithm amp antibodies apparmor argparse automation background backup bash batch-processing binarycif bioinformatics biopython blockchain boltz bootstrapping boxes bubblewrap c-index cca cdn checkpoint cli cloudflare codex cofoldarena concurrency configparser configuration copy cprofile cpu monitoring cron csv cuda curl data-leakage data-processing data-quality data-validation datascience datavisualization datetime debugging decimal deployment desktop-app devtools difflib disk-space disown dns docker dovecot download economics electron encoding esm esm2 esm3 esmc esmfold2 etag evaluation fable faceswap fasta fastmcp faulthandler ffmpeg file-descriptors file-io file-management filecmp filenames files flashppi flask folium frontend game generator git github-actions google google-research gpt-6.1-sol grep gzip harness heapq hls html http hugo imports indexnow ipaddress jailbreak javascript jev json jsonl just k-means kaggle langfuse leecode linux list litellm llm llms.txt logging logs lollipop lsof m3u8 machine-learning macos manacher matplotlib mcp memory mirror mmcif model-evaluation monitoring mp3 mp4 mpnn multiomics multiprocessing mutation mysql nanobert nanobodies networking networkx nginx normalize numpy ollama omegatherm pandas password pathlib pdb pep-723 performance permissions phaser pillow pip postfix preprocessing print probability prompt-engineering protein-alignment protein-design protein-embeddings protein-interactions protein-language-models protein-stability protein-structure proxy pydantic pyecharts pyqt python python3 r raincloud reproducibility requests reservoir-sampling rfoptimization rg ripgrep rosettafold3 roundcube rrsi rsi rsync s-tui sampling scale scikit-learn scrapy screen seaborn security seo sequence-identity sha256 shell signals sitemap sklearn socket solana somaticsignatures sorting spl sql sqlite ssh standardize static-site subprocess symlink sysadmin system-one tarfile tcp tempfile tensorflow tkinter tls toml totp tracemalloc trade tron tronpy troubleshooting turtle typesafe-ai ubuntu unicode upsert urllib usdt utf-8 uv venv vhhbert vite wallet web-tools webp wordcloud wordpress workflow xml yaml zipfile zoneinfo 迅雷解析

友情链接