d3-log-p95--anthropic-claude-fable-5-1

Write a Python CLI `p95paths`: reads an access-log CSV with columns timestamp,path,latency_ms and prints the top-N paths by 95th-percentile latency (default N=10), highest first, as `path p95_ms count`. Use argparse for --top and the input file (stdin when omitted). Malformed rows are skipped with a warning count on stderr; an empty or header-only input prints nothing and exits 0. Deliver a complete, runnable script plus focused tests covering the percentile math and malformed-row handling. Accompany the code with a brief explanation of your design choices.

Prompt

Write a Python CLI `p95paths`: reads an access-log CSV with columns timestamp,path,latency_ms and prints the top-N paths by 95th-percentile latency (default N=10), highest first, as `path p95_ms count`. Use argparse for --top and the input file (stdin when omitted). Malformed rows are skipped with a warning count on stderr; an empty or header-only input prints nothing and exits 0. Deliver a complete, runnable script plus focused tests covering the percentile math and malformed-row handling. Accompany the code with a brief explanation of your design choices. ## Output contract First, write a brief explanation of your design choices — plain prose, no code fences. Then emit exactly one fenced code block per file, and no other text between or after the blocks: - The FIRST line of every block is that file's marker line: - TypeScript or JavaScript: `// FILE: <path>` - Python: `# FILE: <path>` - A single-file web app is one file named `index.html`, so its marker line is `// FILE: index.html`. - Everything after the marker line is that file complete, from its first line to its last — no diffs, no omissions, no placeholders. - Tag each fence with the file's language, for example ```html, ```typescript, ```python.

Artifact

Form: tree

p95paths.py

#!/usr/bin/env python3
"""p95paths: report the top-N request paths by 95th-percentile latency.

Input is a CSV with a header containing at least the columns
``path`` and ``latency_ms`` (a ``timestamp`` column is expected but ignored).
Output lines have the form ``path  p95_ms  count``, highest p95 first.
"""

from __future__ import annotations

import argparse
import csv
import math
import sys
from collections import defaultdict
from typing import Dict, Iterable, List, Optional, Sequence, TextIO, Tuple

REQUIRED_COLUMNS = ("path", "latency_ms")


def percentile(values: Sequence[float], pct: int = 95) -> float:
    """Nearest-rank percentile.

    Returns the value at rank ``ceil(pct/100 * n)`` (1-based) in the sorted
    sample. Uses integer arithmetic so boundary cases are exact.
    Raises ``ValueError`` on an empty sample.
    """
    n = len(values)
    if n == 0:
        raise ValueError("percentile of empty sample")
    if not 0 < pct <= 100:
        raise ValueError("pct must be in (0, 100]")
    rank = (pct * n + 99) // 100  # ceil(pct * n / 100)
    return sorted(values)[rank - 1]


def parse_latency(raw: Optional[str]) -> Optional[float]:
    """Return a finite, non-negative latency or ``None`` if malformed."""
    if raw is None:
        return None
    text = raw.strip()
    if not text:
        return None
    try:
        value = float(text)
    except ValueError:
        return None
    if math.isnan(value) or math.isinf(value) or value < 0:
        return None
    return value


def aggregate(rows: Iterable[Dict[str, Optional[str]]]) -> Tuple[Dict[str, List[float]], int]:
    """Group latencies by path. Returns (latencies_by_path, skipped_count)."""
    by_path: Dict[str, List[float]] = defaultdict(list)
    skipped = 0
    for row in rows:
        # DictReader signals extra fields under the ``None`` key and missing
        # fields as ``None`` values; both mean the row has the wrong shape.
        if None in row:
            skipped += 1
            continue
        path = row.get("path")
        latency = parse_latency(row.get("latency_ms"))
        if path is None or not path.strip() or latency is None:
            skipped += 1
            continue
        by_path[path.strip()].append(latency)
    return by_path, skipped


def rank_paths(by_path: Dict[str, List[float]], top: int) -> List[Tuple[str, float, int]]:
    """Return up to ``top`` (path, p95, count) tuples, highest p95 first."""
    stats = [(path, percentile(vals), len(vals)) for path, vals in by_path.items()]
    stats.sort(key=lambda s: (-s[1], -s[2], s[0]))
    return stats[:top] if top >= 0 else stats


def format_number(value: float) -> str:
    """Render integral values without a trailing ``.0``."""
    if value == int(value):
        return str(int(value))
    return repr(value)


def format_report(stats: Iterable[Tuple[str, float, int]]) -> str:
    return "".join(f"{path}  {format_number(p95)}  {count}\n" for path, p95, count in stats)


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="p95paths",
        description="Print the top-N paths by 95th-percentile latency from an access-log CSV.",
    )
    parser.add_argument(
        "input",
        nargs="?",
        help="CSV file with columns timestamp,path,latency_ms (default: stdin)",
    )
    parser.add_argument(
        "--top",
        "-n",
        type=int,
        default=10,
        help="number of paths to show (default: 10)",
    )
    return parser


def run(stream: TextIO, top: int, stdout: TextIO, stderr: TextIO) -> int:
    reader = csv.DictReader(stream)
    fieldnames = reader.fieldnames
    if fieldnames is None:
        return 0  # completely empty input: nothing to report
    missing = [c for c in REQUIRED_COLUMNS if c not in fieldnames]
    if missing:
        stderr.write(f"p95paths: error: header is missing column(s): {', '.join(missing)}\n")
        return 2
    by_path, skipped = aggregate(reader)
    stdout.write(format_report(rank_paths(by_path, top)))
    if skipped:
        stderr.write(f"p95paths: warning: skipped {skipped} malformed row(s)\n")
    return 0


def main(
    argv: Optional[Sequence[str]] = None,
    stdin: Optional[TextIO] = None,
    stdout: Optional[TextIO] = None,
    stderr: Optional[TextIO] = None,
) -> int:
    stdin = stdin if stdin is not None else sys.stdin
    stdout = stdout if stdout is not None else sys.stdout
    stderr = stderr if stderr is not None else sys.stderr

    parser = build_parser()
    args = parser.parse_args(argv)
    if args.top < 0:
        parser.error("--top must be non-negative")

    if args.input is None:
        return run(stdin, args.top, stdout, stderr)
    try:
        with open(args.input, newline="", encoding="utf-8") as fh:
            return run(fh, args.top, stdout, stderr)
    except OSError as exc:
        stderr.write(f"p95paths: error: cannot read {args.input!r}: {exc.strerror}\n")
        return 1


if __name__ == "__main__":
    sys.exit(main())

test_p95paths.py

import io

import pytest

import p95paths
from p95paths import aggregate, main, parse_latency, percentile, rank_paths


# --- percentile math -------------------------------------------------------

def test_percentile_single_value():
    assert percentile([42.0]) == 42.0


def test_percentile_empty_raises():
    with pytest.raises(ValueError):
        percentile([])


def test_percentile_nearest_rank_n20():
    # ceil(0.95 * 20) = 19 -> 19th smallest of 1..20 is 19
    assert percentile(list(range(1, 21))) == 19


def test_percentile_nearest_rank_n100():
    # ceil(0.95 * 100) = 95 -> 95th smallest of 1..100 is 95
    assert percentile(list(range(1, 101))) == 95


def test_percentile_nearest_rank_n101():
    # ceil(0.95 * 101) = ceil(95.95) = 96
    assert percentile(list(range(1, 102))) == 96


def test_percentile_is_order_independent():
    shuffled = [5, 1, 4, 2, 3]
    assert percentile(shuffled) == percentile(sorted(shuffled))


def test_percentile_small_samples_return_max():
    # For n <= 19, ceil(0.95*n) == n, so p95 is the maximum.
    for n in range(1, 20):
        assert percentile(list(range(n))) == n - 1


def test_percentile_returns_observed_value_not_interpolation():
    assert percentile([10, 1000]) == 1000
    assert percentile([1, 2, 3, 100]) == 100


def test_percentile_other_pct():
    assert percentile(list(range(1, 11)), pct=50) == 5
    assert percentile(list(range(1, 11)), pct=100) == 10


# --- latency parsing -------------------------------------------------------

@pytest.mark.parametrize("raw", ["12", " 12.5 ", "0", "1e3"])
def test_parse_latency_valid(raw):
    assert parse_latency(raw) == float(raw)


@pytest.mark.parametrize("raw", [None, "", "  ", "abc", "-1", "nan", "inf", "12ms"])
def test_parse_latency_malformed(raw):
    assert parse_latency(raw) is None


# --- aggregation & malformed rows -----------------------------------------

def _rows(text):
    import csv
    return csv.DictReader(io.StringIO(text))


def test_aggregate_skips_malformed_rows_and_counts_them():
    text = (
        "timestamp,path,latency_ms\n"
        "t1,/a,10\n"
        "t2,/a,notanumber\n"      # bad latency
        "t3,,15\n"                # empty path
        "t4,/b\n"                 # too few fields
        "t5,/b,20,extra\n"        # too many fields
        "t6,/b,-5\n"              # negative latency
        "t7,/b,30\n"
    )
    by_path, skipped = aggregate(_rows(text))
    assert skipped == 5
    assert by_path == {"/a": [10.0], "/b": [30.0]}


def test_aggregate_accepts_columns_in_any_order():
    text = "latency_ms,path,timestamp\n5,/x,t1\n7,/x,t2\n"
    by_path, skipped = aggregate(_rows(text))
    assert skipped == 0
    assert by_path == {"/x": [5.0, 7.0]}


# --- ranking ---------------------------------------------------------------

def test_rank_paths_orders_by_p95_desc_then_count_desc_then_path():
    by_path = {
        "/slow": [100, 200],
        "/fast": [1, 2, 3],
        "/tie-few": [50],
        "/tie-many": [50, 50],
    }
    ranked = rank_paths(by_path, top=10)
    assert [r[0] for r in ranked] == ["/slow", "/tie-many", "/tie-few", "/fast"]


def test_rank_paths_respects_top():
    by_path = {f"/p{i}": [i] for i in range(10)}
    ranked = rank_paths(by_path, top=3)
    assert [r[0] for r in ranked] == ["/p9", "/p8", "/p7"]


# --- CLI end to end --------------------------------------------------------

def _run(text, argv=()):
    out, err = io.StringIO(), io.StringIO()
    code = main(list(argv), stdin=io.StringIO(text), stdout=out, stderr=err)
    return code, out.getvalue(), err.getvalue()


def test_cli_empty_input_prints_nothing_exit_zero():
    code, out, err = _run("")
    assert code == 0 and out == "" and err == ""


def test_cli_header_only_prints_nothing_exit_zero():
    code, out, err = _run("timestamp,path,latency_ms\n")
    assert code == 0 and out == "" and err == ""


def test_cli_basic_report_format():
    text = (
        "timestamp,path,latency_ms\n"
        "t,/a,10\n"
        "t,/a,20\n"
        "t,/b,5\n"
    )
    code, out, err = _run(text)
    assert code == 0
    assert out == "/a  20  2\n/b  5  1\n"
    assert err == ""


def test_cli_warns_once_on_malformed_rows():
    text = (
        "timestamp,path,latency_ms\n"
        "t,/a,10\n"
        "t,/a,bad\n"
        "t,/a\n"
    )
    code, out, err = _run(text)
    assert code == 0
    assert out == "/a  10  1\n"
    assert "skipped 2 malformed row(s)" in err


def test_cli_top_option_limits_output():
    text = "timestamp,path,latency_ms\n" + "".join(f"t,/p{i},{i}\n" for i in range(5))
    code, out, _ = _run(text, ["--top", "2"])
    assert code == 0
    assert out == "/p4  4  1\n/p3  3  1\n"


def test_cli_missing_required_column_is_error():
    code, out, err = _run("timestamp,latency_ms\nt,5\n")
    assert code == 2
    assert out == ""
    assert "missing column" in err


def test_cli_reads_from_file(tmp_path):
    f = tmp_path / "log.csv"
    f.write_text("timestamp,path,latency_ms\nt,/f,3.5\n", encoding="utf-8")
    out, err = io.StringIO(), io.StringIO()
    code = main([str(f)], stdout=out, stderr=err)
    assert code == 0
    assert out.getvalue() == "/f  3.5  1\n"


def test_cli_missing_file_reports_error():
    out, err = io.StringIO(), io.StringIO()
    code = main(["/nonexistent/definitely/missing.csv"], stdout=out, stderr=err)
    assert code == 1
    assert "cannot read" in err.getvalue()

Explanation

The script uses only the standard library so it runs anywhere. Percentile math is the nearest-rank method: sort each path's latencies and take the sample at position `ceil(0.95·n)` (computed with integer arithmetic to dodge floating-point rounding at boundaries like n=20). This is deterministic, always returns an observed value (so the report never shows a latency nobody actually experienced), and degrades sensibly to the max for tiny samples. It is isolated in a pure `percentile` function so it can be tested exhaustively without touching I/O. Parsing uses `csv.DictReader` so the columns may appear in any order; a header lacking `path` or `latency_ms` is a fatal usage error (exit 2), whereas a completely empty stream simply produces no output and exit 0. A row is "malformed" if it has the wrong field count, an empty path, or a latency that is not a finite non-negative number; such rows are counted and one summary warning goes to stderr, keeping stdout clean for piping. Latencies are accumulated per path in memory, which is fine for typical log sizes and keeps the p95 exact rather than approximated. Output is sorted by p95 descending, then count descending, then path ascending, so ties are stable and reproducible. `main` accepts injectable streams so the tests can drive the CLI end-to-end with `StringIO` without spawning subprocesses.

Judge detail

No raw judge output is published for this item yet. When it is, it lands under results/raw/ and appears here verbatim; the rubric and protocol are already documented on the Methodology page.