Parsing Logs & Text

Ops scripts extract signals from log files and command output using streaming I/O, regex, and optional JSON parsing - without loading multi-gigabyte files into memory.

Recipe

import re
from pathlib import Path
 
PATTERN = re.compile(r"ERROR (?P<component>\w+): (?P<message>.+)")
 
for line in Path("app.log").open(encoding="utf-8", errors="replace"):
    match = PATTERN.search(line)
    if match:
        print(match.groupdict())

When to reach for this:

• Incident triage - grep-like scripts with richer logic
• Metrics from logs - count error codes per hour
• Ad-hoc ETL from legacy text formats
• Validate export files before upload

Working Example

Stream nginx access log, aggregate status codes, handle JSON lines mixed format.

import json
import re
from collections import Counter
from dataclasses import dataclass
from pathlib import Path
 
NGINX = re.compile(
    r'^(?P<ip>\S+) \S+ \S+ \[(?P<time>[^\]]+)\] "(?P<method>\w+) (?P<path>\S+)'
)
 
@dataclass
class ParseResult:
    status_counts: Counter
    errors: int
 
def parse_access_log(path: Path) -> ParseResult:
    counts: Counter = Counter()
    errors = 0
    with path.open(encoding="utf-8", errors="replace") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            if line.startswith("{"):
                try:
                    rec = json.loads(line)
                    counts[rec.get("status", "unknown")] += 1
                    continue
                except json.JSONDecodeError:
                    errors += 1
                    continue
            m = NGINX.search(line)
            if not m:
                errors += 1
                continue
            # status code often after request - simplified demo uses path token
            counts["parsed"] += 1
    return ParseResult(counts, errors)
 
if __name__ == "__main__":
    result = parse_access_log(Path("access.log"))
    print(result.status_counts)
    print("parse errors", result.errors)

What this demonstrates:

• Line-by-line streaming keeps memory flat
• errors="replace" avoids crash on invalid UTF-8 bytes
• JSON lines path supports structured logs alongside legacy text

Deep Dive

Parsing Strategy

1. Try structured parse (JSON) first
2. Fall back to regex for legacy lines
3. Count/unparseable lines for data quality metrics

Regex Tips

• Use named groups (?P<name>...) for readable extraction
• Compile patterns once outside loops
• Anchor carefully - log formats vary by version

Python Notes

import gzip
from pathlib import Path
 
with gzip.open(Path("app.log.gz"), "rt", encoding="utf-8", errors="replace") as f:
    for line in f:
        ...

Gotchas

• Reading entire file with .read() - OOM on GB logs. Fix: iterate lines or use mmap for advanced cases.
• Greedy regex on variable fields - catastrophic backtracking. Fix: possessive quantifiers or split fields without regex where possible.
• Assuming single log format - upgrades break parser. Fix: version detection or permissive multi-parser with error counter.
• Local timezone in timestamps - wrong aggregation buckets. Fix: parse to UTC with datetime.timezone.
• Logging parsed PII - GDPR incident. Fix: aggregate counts, redact IPs in output.

Alternatives

FAQs

Related

• File & Directory Automation - log file paths
• Structured Logging - JSON log production
• Robust Scripts - exit codes on parse failure thresholds
• subprocess & Shell Interop - parse command output
• System Scripting Best Practices - ops script hygiene

Stack versions: This page was written for Python 3.14.0 (stable 3.14, maintenance 3.13), FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+, and uv 0.6+.