Эталонное решениеСначала выполните проект самостоятельно и используйте этот материал для сверки решений.

Проект 1. Эталонное решение

Открывать после собственной реализации и прохождения CHECKLIST.md. Прочитанное решение всегда кажется очевидным — это свойство чтения, а не признак понимания.

Решение — не единственно верное. Места, где обоснован другой выбор, отмечены явно.

Что проверено на самом деле

Что	Как	Результат
Код приложения	фактический запуск	работает
Тесты	`pytest`	48 пройдено
Покрытие	`pytest-cov`	93 %
Установка пакета	`pip install .` в чистом окружении	работает
Все четыре кода возврата	запуск с разными входами	различимы
`Dockerfile`	не собирался	Docker на машине курса не установлен

Последняя строка существенна: Dockerfile ниже написан по правилам разделов 05 и 15, но сборка не выполнялась. Проверьте её у себя — CHECKLIST.md содержит нужные команды.

Структура

logstat/
├── Dockerfile
├── .dockerignore
├── pyproject.toml
├── README.md
├── logstat.example.toml
├── src/logstat/
│   ├── __init__.py
│   ├── __main__.py
│   ├── errors.py
│   ├── config.py
│   ├── stats.py
│   └── cli.py
└── tests/
    ├── test_config.py
    ├── test_stats.py
    └── test_cli.py

Порядок изложения ниже — снизу вверх: от того, что ни от чего не зависит, к оболочке.

Шаг 1. Коды возврата как часть модели ошибок

Первое решение, определяющее остальные: код возврата прикреплён к исключению, а не выбирается в месте обработки.

src/logstat/errors.py:

"""Коды возврата и исключения.

Три различимых кода вместо двух — сознательное решение: «записей
не найдено» и «произошла ошибка» требуют разной реакции в скриптах.
"""
from __future__ import annotations

EXIT_OK = 0
EXIT_RUNTIME = 1
EXIT_USAGE = 2
EXIT_NO_MATCH = 3


class LogstatError(Exception):
    """Базовая ошибка приложения."""

    exit_code = EXIT_RUNTIME


class UsageError(LogstatError):
    """Неверное использование: аргументы или конфигурация."""

    exit_code = EXIT_USAGE


class RuntimeFailure(LogstatError):
    """Ошибка выполнения: недоступный файл, испорченные данные."""

    exit_code = EXIT_RUNTIME

Альтернатива — возвращать коды из функций — приводит к тому, что каждый вызывающий обязан их пробрасывать, и один пропущенный return превращает ошибку в успех. Здесь достаточно одного обработчика в main().

Где возможен другой выбор. Различать RuntimeFailure и базовый LogstatError строго необязательно: сейчас у них один код. Разделение оставлено как место для роста — например, для кода 4 «частичный результат».

Шаг 2. Конфигурация из трёх источников

src/logstat/config.py:

"""Конфигурация: файл TOML, переменные окружения, аргументы.

Приоритет — от низшего к высшему: значения по умолчанию, файл,
переменные окружения, аргументы командной строки. Источник каждого
значения запоминается: без этого невозможно ответить на вопрос
«почему применилось не то, что я задал».
"""
from __future__ import annotations

import os
import tomllib
from dataclasses import dataclass, field
from pathlib import Path

from .errors import UsageError

ENV_PREFIX = "LOGSTAT_"
DEFAULT_CONFIG_PATH = Path("/etc/logstat/config.toml")

FORMATS = ("text", "json")


@dataclass
class Config:
    """Итоговая конфигурация и происхождение каждого значения."""

    top: int = 10
    group_by: str = "status"
    output_format: str = "text"
    strict: bool = False
    sources: dict[str, str] = field(default_factory=dict)

    def describe(self) -> list[tuple[str, object, str]]:
        return [
            ("top", self.top, self.sources.get("top", "умолчание")),
            ("group_by", self.group_by, self.sources.get("group_by", "умолчание")),
            ("output_format", self.output_format,
             self.sources.get("output_format", "умолчание")),
            ("strict", self.strict, self.sources.get("strict", "умолчание")),
        ]


def _coerce(key: str, raw: object, origin: str) -> object:
    """Приведение к типу с понятным сообщением об ошибке."""
    if key == "top":
        try:
            value = int(raw)  # type: ignore[arg-type]
        except (TypeError, ValueError):
            raise UsageError(f"top: ожидалось целое число, получено {raw!r} ({origin})")
        if value < 1:
            raise UsageError(f"top: ожидалось положительное число, получено {value} ({origin})")
        return value
    if key == "output_format":
        value = str(raw)
        if value not in FORMATS:
            raise UsageError(
                f"output_format: ожидалось одно из {', '.join(FORMATS)}, "
                f"получено {value!r} ({origin})")
        return value
    if key == "strict":
        if isinstance(raw, bool):
            return raw
        value = str(raw).strip().lower()
        if value in ("1", "true", "yes", "on"):
            return True
        if value in ("0", "false", "no", "off"):
            return False
        raise UsageError(f"strict: ожидалось логическое значение, получено {raw!r} ({origin})")
    return str(raw)


def _from_file(path: Path) -> dict[str, object]:
    try:
        data = tomllib.loads(path.read_text(encoding="utf-8"))
    except FileNotFoundError:
        raise UsageError(f"файл конфигурации не найден: {path}")
    except tomllib.TOMLDecodeError as exc:
        raise UsageError(f"файл конфигурации испорчен: {path}: {exc}")
    section = data.get("logstat", data)
    if not isinstance(section, dict):
        raise UsageError(f"файл конфигурации: ожидалась таблица [logstat] в {path}")
    return section


def _from_env(env: dict[str, str]) -> dict[str, object]:
    out: dict[str, object] = {}
    for name, value in env.items():
        if not name.startswith(ENV_PREFIX):
            continue
        key = name[len(ENV_PREFIX):].lower()
        out[key] = value
    return out


def load(
    config_path: Path | None,
    overrides: dict[str, object],
    env: dict[str, str] | None = None,
) -> Config:
    """Собрать конфигурацию из всех источников по приоритету."""
    env = os.environ if env is None else env
    cfg = Config()
    known = {"top", "group_by", "output_format", "strict"}

    layers: list[tuple[str, dict[str, object]]] = []

    path = config_path
    if path is None:
        env_path = env.get(f"{ENV_PREFIX}CONFIG")
        if env_path:
            path = Path(env_path)
        elif DEFAULT_CONFIG_PATH.exists():
            path = DEFAULT_CONFIG_PATH
    if path is not None:
        layers.append((f"файл {path}", _from_file(path)))

    layers.append(("окружение", _from_env(env)))
    layers.append(("аргумент", {k: v for k, v in overrides.items() if v is not None}))

    for origin, values in layers:
        for key, raw in values.items():
            if key == "config":
                continue
            if key not in known:
                raise UsageError(f"неизвестный параметр {key!r} ({origin})")
            setattr(cfg, key, _coerce(key, raw, origin))
            cfg.sources[key] = origin

    return cfg

Три решения, которые стоит разобрать.

Источник каждого значения запоминается. Поле sources и метод describe() существуют ради одного вопроса: «почему применилось не то, что я задал». Без них ответ ищут перебором. Это то же требование к диагностируемости, что healthcheck предъявляет к сервису.

Отсутствующий аргумент отфильтровывается явно. Строка

layers.append(("аргумент", {k: v for k, v in overrides.items() if v is not None}))

решает частую ошибку: argparse кладёт None в неуказанные параметры, и наивное dict.update() затрёт значение из окружения. В тестах это test_none_argument_does_not_override.

Неизвестный параметр — ошибка, а не молчаливое игнорирование. LOGSTAT_TPO=5 (опечатка) должен приводить к отказу. Молчаливое игнорирование даёт худший исход: программа работает и делает не то.

Где возможен другой выбор. Приведение типов написано вручную. Для более крупного проекта разумнее pydantic-settings (урок 6.5); здесь зависимость не добавлена намеренно, чтобы образ собирался без единого стороннего пакета.

Шаг 3. Подсчёт

src/logstat/stats.py:

"""Разбор записей и подсчёт статистики."""
from __future__ import annotations

import json
from collections import Counter
from dataclasses import dataclass
from typing import Iterable, Iterator

from .errors import RuntimeFailure


@dataclass
class Skipped:
    """Строки, которые не удалось разобрать."""

    count: int = 0
    first_line: int | None = None
    first_reason: str = ""

    def add(self, line_no: int, reason: str) -> None:
        self.count += 1
        if self.first_line is None:
            self.first_line = line_no
            self.first_reason = reason


@dataclass
class Report:
    """Результат подсчёта."""

    group_by: str
    total: int
    matched: int
    counts: Counter[str]
    skipped: Skipped

    def top(self, n: int) -> list[tuple[str, int]]:
        # most_common сортирует по убыванию счётчика; при равенстве
        # порядок вставки не воспроизводим между запусками, поэтому
        # добавлена вторая ось сортировки — по ключу.
        return sorted(self.counts.items(), key=lambda kv: (-kv[1], kv[0]))[:n]


def parse_filter(expr: str | None) -> tuple[str, str] | None:
    """Разобрать выражение вида поле=значение."""
    if expr is None:
        return None
    if "=" not in expr:
        raise RuntimeFailure(f"фильтр должен иметь вид поле=значение, получено {expr!r}")
    field, _, value = expr.partition("=")
    field = field.strip()
    if not field:
        raise RuntimeFailure(f"фильтр без имени поля: {expr!r}")
    return field, value.strip()


def read_records(lines: Iterable[str], strict: bool,
                 skipped: Skipped) -> Iterator[dict[str, object]]:
    """Разобрать JSON Lines, пропуская или отвергая испорченные строки."""
    for line_no, raw in enumerate(lines, 1):
        line = raw.strip()
        if not line:
            continue
        try:
            record = json.loads(line)
        except json.JSONDecodeError as exc:
            reason = f"строка {line_no}: не JSON ({exc.msg})"
            if strict:
                raise RuntimeFailure(reason)
            skipped.add(line_no, reason)
            continue
        if not isinstance(record, dict):
            reason = f"строка {line_no}: ожидался объект, получен {type(record).__name__}"
            if strict:
                raise RuntimeFailure(reason)
            skipped.add(line_no, reason)
            continue
        yield record


def build_report(lines: Iterable[str], group_by: str,
                 filter_expr: str | None, strict: bool) -> Report:
    """Посчитать статистику по потоку строк."""
    condition = parse_filter(filter_expr)
    skipped = Skipped()
    counts: Counter[str] = Counter()
    total = matched = 0

    for record in read_records(lines, strict, skipped):
        total += 1
        if condition is not None:
            field, expected = condition
            if str(record.get(field, "")) != expected:
                continue
        matched += 1
        counts[str(record.get(group_by, "—"))] += 1

    return Report(group_by=group_by, total=total, matched=matched,
                  counts=counts, skipped=skipped)

Решение о порядке сортировки. Counter.most_common() при равных счётчиках возвращает элементы в порядке вставки — то есть результат зависит от порядка строк во входном файле. Для инструмента, чей вывод сравнивают между запусками, это плохо. Вторая ось сортировки по ключу делает вывод воспроизводимым; тест test_top_is_deterministic_on_ties это фиксирует.

Решение о неразобранных строках. По умолчанию строка пропускается, но считается, и число попадает в диагностику. Молчаливый пропуск дал бы правдоподобный неверный ответ — исход хуже отказа. Режим --strict переключает поведение на отказ, что нужно в конвейере, где данные обязаны быть чистыми.

Шаг 4. Оболочка командной строки

src/logstat/cli.py:

"""Точка входа командной строки."""
from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path
from typing import IO, Sequence

from . import __version__
from .config import Config, load
from .errors import EXIT_NO_MATCH, EXIT_OK, LogstatError, RuntimeFailure, UsageError
from .stats import Report, build_report


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="logstat",
        description="Статистика по журналам в формате JSON Lines.",
        epilog="Коды возврата: 0 успех, 1 ошибка выполнения, "
               "2 ошибка использования, 3 записей не найдено.",
    )
    parser.add_argument("path", nargs="?", default="-",
                        help="файл журнала; «-» или пропуск — стандартный ввод")
    parser.add_argument("--group-by", dest="group_by", metavar="ПОЛЕ",
                        help="поле для группировки")
    parser.add_argument("--top", type=int, metavar="N",
                        help="сколько строк показать")
    parser.add_argument("--filter", dest="filter_expr", metavar="ПОЛЕ=ЗНАЧЕНИЕ",
                        help="оставить записи с указанным значением поля")
    parser.add_argument("--format", dest="output_format", choices=("text", "json"),
                        help="формат вывода")
    parser.add_argument("--strict", action="store_true", default=None,
                        help="прерываться на первой неразобранной строке")
    parser.add_argument("--config", type=Path, metavar="ПУТЬ",
                        help="файл конфигурации TOML")
    parser.add_argument("--show-config", action="store_true",
                        help="показать итоговую конфигурацию и источник каждого значения")
    parser.add_argument("--version", action="version",
                        version=f"logstat {__version__}")
    return parser


def open_source(path: str, stdin: IO[str]) -> tuple[IO[str], bool]:
    """Вернуть поток строк и признак, нужно ли его закрывать."""
    if path == "-":
        return stdin, False
    p = Path(path)
    try:
        return p.open(encoding="utf-8"), True
    except FileNotFoundError:
        raise RuntimeFailure(f"файл не найден: {path}")
    except IsADirectoryError:
        raise RuntimeFailure(f"это каталог, а не файл: {path}")
    except PermissionError:
        raise RuntimeFailure(f"нет прав на чтение: {path}")


def render_text(report: Report, top: int, out: IO[str]) -> None:
    rows = report.top(top)
    width = max((len(k) for k, _ in rows), default=len(report.group_by))
    width = max(width, len(report.group_by))
    print(f"{report.group_by:<{width}}  {'записей':>8}  {'доля':>7}", file=out)
    print("─" * (width + 19), file=out)
    for key, count in rows:
        share = count / report.matched * 100 if report.matched else 0.0
        print(f"{key:<{width}}  {count:>8}  {share:>6.1f}%", file=out)


def render_json(report: Report, top: int, out: IO[str]) -> None:
    payload = {
        "группировка": report.group_by,
        "всего": report.total,
        "отобрано": report.matched,
        "пропущено": report.skipped.count,
        "строки": [{"ключ": k, "записей": n} for k, n in report.top(top)],
    }
    json.dump(payload, out, ensure_ascii=False, indent=2)
    out.write("\n")


def report_diagnostics(report: Report, cfg: Config, err: IO[str]) -> None:
    """Диагностика идёт в stderr, чтобы не смешиваться с результатом."""
    print(f"прочитано записей: {report.total}, отобрано: {report.matched}", file=err)
    if report.skipped.count:
        print(f"пропущено неразобранных строк: {report.skipped.count} "
              f"(первая — {report.skipped.first_reason})", file=err)
        print("подсказка: --strict прерывает работу на первой такой строке", file=err)


def show_config(cfg: Config, out: IO[str]) -> None:
    print(f"{'параметр':<16} {'значение':<12} источник", file=out)
    print("─" * 48, file=out)
    for name, value, origin in cfg.describe():
        print(f"{name:<16} {str(value):<12} {origin}", file=out)


def main(argv: Sequence[str] | None = None,
         stdin: IO[str] | None = None,
         stdout: IO[str] | None = None,
         stderr: IO[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    stdin = stdin or sys.stdin
    out = stdout or sys.stdout
    err = stderr or sys.stderr

    try:
        cfg = load(args.config, {
            "top": args.top,
            "group_by": args.group_by,
            "output_format": args.output_format,
            "strict": args.strict,
        })

        if args.show_config:
            show_config(cfg, out)
            return EXIT_OK

        source, need_close = open_source(args.path, stdin)
        try:
            report = build_report(source, cfg.group_by, args.filter_expr, cfg.strict)
        finally:
            if need_close:
                source.close()

        if cfg.output_format == "json":
            render_json(report, cfg.top, out)
        else:
            render_text(report, cfg.top, out)

        report_diagnostics(report, cfg, err)

        if report.matched == 0:
            print("записей, удовлетворяющих условию, не найдено", file=err)
            return EXIT_NO_MATCH
        return EXIT_OK

    except LogstatError as exc:
        print(f"logstat: {exc}", file=err)
        return exc.exit_code
    except BrokenPipeError:
        # Обычная ситуация при `logstat ... | head`: получатель закрыл поток.
        return EXIT_OK
    except KeyboardInterrupt:
        print("logstat: прервано", file=err)
        return 130

Четыре решения.

Потоки передаются параметрами, а не берутся из sys. Сигнатура main(argv, stdin, stdout, stderr) позволяет тестировать разделение потоков без подмены глобального состояния. Это сделало возможными тесты test_result_goes_to_stdout_diagnostics_to_stderr и test_json_output_contains_only_json — самые ценные в наборе.

BrokenPipeError перехватывается и даёт код 0. logstat ... | head -2 закрывает поток на середине. Без обработки Python печатает трассировку — то есть штатное использование выглядит как отказ.

Диагностика печатается после результата, но это не гарантирует порядок. При перенаправлении stdout буферизуется блоками, stderr — нет. Смешанный вывод может идти в любом порядке; это отмечено в CHECKLIST.md и не является ошибкой.

--show-config печатает в stdout и завершает работу. Это результат работы, а не диагностика: его разбирают скриптами.

Точка входа:

src/logstat/__init__.py:

"""logstat — статистика по журналам в формате JSON Lines."""

__version__ = "1.0.0"

src/logstat/__main__.py:

"""Запуск через `python -m logstat`."""
from __future__ import annotations

import sys

from .cli import main

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

Наличие __main__.py даёт python -m logstat — форма, работающая без установки консольного скрипта. Полезно при отладке внутри container'а.

Шаг 5. Тесты

Набор разделён по слоям: конфигурация, подсчёт, оболочка. Тесты оболочки — самые содержательные, потому что проверяют именно то, что ломается чаще всего.

tests/test_config.py:

"""Конфигурация: приоритет источников и сообщения об ошибках."""
from __future__ import annotations

import pytest

from logstat.config import load
from logstat.errors import UsageError


def test_defaults_when_nothing_given():
    cfg = load(None, {}, env={})
    assert cfg.top == 10
    assert cfg.group_by == "status"
    assert cfg.output_format == "text"
    assert cfg.strict is False


def test_env_overrides_defaults():
    cfg = load(None, {}, env={"LOGSTAT_TOP": "3"})
    assert cfg.top == 3
    assert cfg.sources["top"] == "окружение"


def test_argument_overrides_env():
    cfg = load(None, {"top": 5}, env={"LOGSTAT_TOP": "3"})
    assert cfg.top == 5
    assert cfg.sources["top"] == "аргумент"


def test_file_is_lowest_of_the_three(tmp_path):
    path = tmp_path / "c.toml"
    path.write_text('[logstat]\ntop = 7\ngroup_by = "path"\n', encoding="utf-8")
    cfg = load(path, {}, env={"LOGSTAT_TOP": "3"})
    assert cfg.top == 3, "окружение должно перекрывать файл"
    assert cfg.group_by == "path", "файл применяется там, где нет других источников"
    assert cfg.sources["group_by"].startswith("файл")


def test_none_argument_does_not_override():
    """Отсутствующий аргумент — не то же самое, что заданный пустым."""
    cfg = load(None, {"top": None}, env={"LOGSTAT_TOP": "3"})
    assert cfg.top == 3


@pytest.mark.parametrize("value", ["zero", "-1", "0", ""])
def test_bad_top_is_usage_error(value):
    with pytest.raises(UsageError) as exc:
        load(None, {}, env={"LOGSTAT_TOP": value})
    assert "top" in str(exc.value)


def test_bad_format_names_allowed_values():
    with pytest.raises(UsageError) as exc:
        load(None, {}, env={"LOGSTAT_OUTPUT_FORMAT": "xml"})
    assert "text" in str(exc.value) and "json" in str(exc.value)


def test_unknown_key_is_rejected():
    with pytest.raises(UsageError) as exc:
        load(None, {}, env={"LOGSTAT_NOSUCH": "1"})
    assert "nosuch" in str(exc.value)


def test_missing_config_file_is_usage_error(tmp_path):
    with pytest.raises(UsageError):
        load(tmp_path / "нет.toml", {}, env={})


def test_broken_config_file_names_the_path(tmp_path):
    path = tmp_path / "c.toml"
    path.write_text("[logstat\ntop = 1", encoding="utf-8")
    with pytest.raises(UsageError) as exc:
        load(path, {}, env={})
    assert str(path) in str(exc.value)


@pytest.mark.parametrize("raw,expected", [
    ("1", True), ("true", True), ("yes", True), ("on", True),
    ("0", False), ("false", False), ("no", False), ("off", False),
])
def test_strict_accepts_usual_spellings(raw, expected):
    cfg = load(None, {}, env={"LOGSTAT_STRICT": raw})
    assert cfg.strict is expected

tests/test_stats.py:

"""Разбор записей и подсчёт."""
from __future__ import annotations

import pytest

from logstat.errors import RuntimeFailure
from logstat.stats import build_report, parse_filter

LINES = [
    '{"status": 200, "path": "/a"}',
    '{"status": 200, "path": "/b"}',
    '{"status": 404, "path": "/a"}',
    "",
    "не json",
    "[1, 2]",
]


def test_counts_by_field():
    r = build_report(LINES, "status", None, strict=False)
    assert r.total == 3
    assert r.matched == 3
    assert dict(r.counts) == {"200": 2, "404": 1}


def test_skipped_lines_are_counted_not_hidden():
    r = build_report(LINES, "status", None, strict=False)
    assert r.skipped.count == 2
    assert r.skipped.first_line == 5


def test_empty_lines_are_not_errors():
    r = build_report(["", "   ", '{"status": 1}'], "status", None, strict=False)
    assert r.skipped.count == 0
    assert r.total == 1


def test_strict_stops_at_first_bad_line():
    with pytest.raises(RuntimeFailure) as exc:
        build_report(LINES, "status", None, strict=True)
    assert "строка 5" in str(exc.value)


def test_filter_reduces_matched_but_not_total():
    r = build_report(LINES, "status", "path=/a", strict=False)
    assert r.total == 3
    assert r.matched == 2


def test_missing_field_becomes_placeholder():
    r = build_report(['{"status": 200}'], "нет_такого", None, strict=False)
    assert dict(r.counts) == {"—": 1}


def test_top_is_deterministic_on_ties():
    """При равных счётчиках порядок задаётся ключом, а не порядком вставки."""
    lines = ['{"k": "b"}', '{"k": "a"}', '{"k": "c"}']
    r = build_report(lines, "k", None, strict=False)
    assert [k for k, _ in r.top(3)] == ["a", "b", "c"]


def test_top_limits_rows():
    lines = [f'{{"k": "{c}"}}' for c in "abcde"]
    r = build_report(lines, "k", None, strict=False)
    assert len(r.top(2)) == 2


@pytest.mark.parametrize("expr", ["без-равно", "=значение"])
def test_bad_filter_is_runtime_failure(expr):
    with pytest.raises(RuntimeFailure):
        parse_filter(expr)


def test_filter_value_may_be_empty():
    assert parse_filter("field=") == ("field", "")

tests/test_cli.py:

"""Поведение командной строки: коды возврата и разделение потоков."""
from __future__ import annotations

import io
import json

import pytest

from logstat.cli import main
from logstat.errors import EXIT_NO_MATCH, EXIT_OK, EXIT_RUNTIME, EXIT_USAGE

SAMPLE = "\n".join([
    '{"status": 200, "path": "/a", "method": "GET"}',
    '{"status": 200, "path": "/b", "method": "GET"}',
    '{"status": 500, "path": "/a", "method": "POST"}',
    "испорченная строка",
])


def run(argv, stdin_text="", env=None, monkeypatch=None):
    out, err = io.StringIO(), io.StringIO()
    if env is not None and monkeypatch is not None:
        for k, v in env.items():
            monkeypatch.setenv(k, v)
    code = main(argv, stdin=io.StringIO(stdin_text), stdout=out, stderr=err)
    return code, out.getvalue(), err.getvalue()


def test_reads_stdin_by_default():
    code, out, _ = run([], SAMPLE)
    assert code == EXIT_OK
    assert "200" in out


def test_result_goes_to_stdout_diagnostics_to_stderr():
    _, out, err = run([], SAMPLE)
    assert "200" in out
    assert "прочитано записей" in err
    assert "прочитано записей" not in out, "диагностика не должна попадать в stdout"


def test_skipped_lines_reported_to_stderr():
    _, out, err = run([], SAMPLE)
    assert "пропущено" in err
    assert "пропущено" not in out


def test_no_matches_returns_dedicated_code():
    code, _, err = run(["--filter", "method=DELETE"], SAMPLE)
    assert code == EXIT_NO_MATCH
    assert "не найдено" in err


def test_missing_file_is_runtime_error():
    code, out, err = run(["/нет/такого/файла.jsonl"])
    assert code == EXIT_RUNTIME
    assert out == "", "при ошибке stdout остаётся пустым"
    assert "не найден" in err


def test_bad_env_value_is_usage_error(monkeypatch):
    code, _, err = run([], SAMPLE, env={"LOGSTAT_TOP": "много"}, monkeypatch=monkeypatch)
    assert code == EXIT_USAGE
    assert "top" in err


def test_strict_turns_skips_into_failure():
    code, out, err = run(["--strict"], SAMPLE)
    assert code == EXIT_RUNTIME
    assert out == ""


def test_json_output_is_valid_json():
    _, out, _ = run(["--format", "json"], SAMPLE)
    payload = json.loads(out)
    assert payload["всего"] == 3
    assert payload["пропущено"] == 1


def test_json_output_contains_only_json():
    """Диагностика в stdout сделала бы вывод неразбираемым."""
    _, out, _ = run(["--format", "json"], SAMPLE)
    json.loads(out)  # упадёт, если в stdout попало что-то ещё


def test_top_limits_rows_in_output():
    lines = "\n".join(f'{{"k": "{c}"}}' for c in "abcdef")
    _, out, _ = run(["--group-by", "k", "--top", "2"], lines)
    body = [l for l in out.splitlines() if l and not l.startswith(("k ", "─"))]
    assert len(body) == 2


def test_show_config_reports_source_of_each_value(monkeypatch):
    code, out, _ = run(["--show-config", "--top", "4"],
                       env={"LOGSTAT_GROUP_BY": "path"}, monkeypatch=monkeypatch)
    assert code == EXIT_OK
    assert "аргумент" in out
    assert "окружение" in out


def test_file_argument_is_read(tmp_path):
    path = tmp_path / "log.jsonl"
    path.write_text(SAMPLE, encoding="utf-8")
    code, out, _ = run([str(path)])
    assert code == EXIT_OK
    assert "200" in out


def test_directory_instead_of_file_is_runtime_error(tmp_path):
    code, _, err = run([str(tmp_path)])
    assert code == EXIT_RUNTIME
    assert "каталог" in err


def test_bad_filter_expression_is_runtime_error():
    code, _, err = run(["--filter", "безравно"], SAMPLE)
    assert code == EXIT_RUNTIME
    assert "поле=значение" in err


def test_version_exits_zero():
    with pytest.raises(SystemExit) as exc:
        main(["--version"])
    assert exc.value.code == 0


def test_unknown_argument_exits_two():
    """argparse сам возвращает 2 — тот же код, что и наша UsageError."""
    with pytest.raises(SystemExit) as exc:
        main(["--нет-такого"])
    assert exc.value.code == EXIT_USAGE

Что в этом наборе главное.

Тест test_result_goes_to_stdout_diagnostics_to_stderr содержит утверждение

assert "прочитано записей" not in out

Именно оно ловит самую частую ошибку — print() без file=sys.stderr. Тест, читающий объединённый вывод, эту ошибку пропустит: строка ведь напечатана.

Тест test_json_output_contains_only_json — то же требование, выраженное сильнее: json.loads() упадёт, если в stdout попало хоть что-то лишнее.

Фактический результат:

48 passed in 0.15s

Name                      Stmts   Miss  Cover   Missing
-------------------------------------------------------
src/logstat/__init__.py       1      0   100%
src/logstat/__main__.py       5      5     0%   2-9
src/logstat/cli.py           89      7    92%   55-56, 142-147
src/logstat/config.py        86      5    94%   68, 81, 111, 113, 123
src/logstat/errors.py        11      0   100%
src/logstat/stats.py         69      1    99%   75
-------------------------------------------------------
TOTAL                       261     18    93%

Что не покрыто и почему это оставлено так. Непокрытые строки — PermissionError при открытии файла, KeyboardInterrupt, BrokenPipeError и несколько ветвей проверки типов в конфигурации. Все они воспроизводятся только через подмену системного поведения, и тест получился бы проверкой заглушки, а не кода. __main__.py не покрыт по устройству: он выполняется при запуске модуля, а не при импорте.

Порог 90 % выбран так, чтобы эти строки укладывались в остаток. Порог 100 % заставил бы писать тесты ради числа — и это отдельный антипаттерн (урок 15.2).

Шаг 6. Упаковка

pyproject.toml:

[project]
name = "logstat"
version = "1.0.0"
description = "Статистика по журналам в формате JSON Lines"
readme = "README.md"
requires-python = ">=3.11"
dependencies = []

[project.scripts]
logstat = "logstat.cli:main"

[build-system]
requires = ["setuptools>=69"]
build-backend = "setuptools.build_meta"

[tool.setuptools.packages.find]
where = ["src"]

[tool.pytest.ini_options]
testpaths = ["tests"]
pythonpath = ["src"]
addopts = "-q"

pythonpath = ["src"] в настройках pytest позволяет запускать тесты без установки пакета — удобно локально. В образе пакет именно устанавливается, поэтому тесты в стадии test проверяют установленный код, а не исходники рядом.

Шаг 7. `Dockerfile`

# syntax=docker/dockerfile:1

# ── Общая основа: одна и та же для сборки и для запуска ──────────────
FROM python:3.13-slim AS base
ENV PYTHONUNBUFFERED=1 \
    PYTHONDONTWRITEBYTECODE=1 \
    PATH=/opt/venv/bin:$PATH
WORKDIR /app

# ── Зависимости и установка пакета ───────────────────────────────────
FROM base AS deps
RUN python -m venv /opt/venv
# Файлы, влияющие на установку, копируются ДО исходников:
# изменение кода не сбрасывает кэш этого слоя.
COPY pyproject.toml README.md ./
COPY src/ ./src/
# Кэш pip живёт в mount, а не в слое: --no-cache-dir здесь был бы
# прямым противоречием — он отключает ровно то, что кэшируется.
RUN --mount=type=cache,target=/root/.cache/pip \
    pip install .

# ── Тесты: отдельная стадия, выполняется только при --target test ────
FROM deps AS test
RUN --mount=type=cache,target=/root/.cache/pip \
    pip install pytest==9.1.1 pytest-cov==7.1.0
COPY tests/ ./tests/
RUN python -m pytest -q --cov=logstat --cov-report=term-missing --cov-fail-under=90

# ── Итоговый образ ───────────────────────────────────────────────────
FROM base AS runtime
COPY --from=deps /opt/venv /opt/venv
USER 10001:10001
ENTRYPOINT ["logstat"]
CMD ["--help"]

Четыре решения.

Виртуальное окружение в /opt/venv и копирование его целиком. Альтернатива — копировать site-packages — требует знать точный путь с версией Python (/usr/local/lib/python3.13/...) и ломается при обновлении базового образа. Фиксированный путь надёжнее.

--mount=type=cache без --no-cache-dir. Их сочетание встречается часто и бессмысленно: --no-cache-dir отключает ровно тот кэш, который монтируется. Ошибка безобидная, но показывает, что строку скопировали не читая.

Стадия test наследует deps, а не base. Тесты выполняются против установленного пакета. Если бы стадия начиналась с base, пришлось бы устанавливать всё заново, и проверялось бы другое дерево зависимостей.

USER 10001:10001 без создания пользователя в /etc/passwd. Числовой идентификатор работает и без записи в passwd; для Kubernetes он обязателен (урок 18.2). Ограничение: у процесса нет $HOME. Для этого приложения безразлично; приложению, пишущему в домашний каталог, пользователя пришлось бы создать.

.dockerignore:

.git
.gitignore
.venv
.v
__pycache__
*.pyc
.pytest_cache
.mypy_cache
.ruff_cache
htmlcov
.coverage
*.jsonl

*.jsonl в списке — не мелочь: без него тестовые журналы попадают в контекст сборки и сбрасывают кэш при каждом изменении данных.

logstat.example.toml:

# Пример конфигурации logstat.
# Значения перекрываются переменными окружения LOGSTAT_* и аргументами.

[logstat]
top = 5
group_by = "path"
output_format = "text"
strict = false

Чего решение не делает

Образ не собран. Docker на машине, где готовился курс, отсутствует. Dockerfile написан по правилам разделов 05, 06 и 15, но ни docker build, ни docker run не выполнялись. Всё остальное — код, тесты, покрытие, коды возврата, приоритет конфигурации — проверено фактическим запуском.

Размер образа не измерен. Дополнительное задание 5 (distroless или alpine) не выполнено. Приблизительная оценка была бы выдумкой: размер зависит от базового образа и версии Python.

Воспроизводимость сборки не проверена. Дополнительное задание 6 требует двух сборок подряд и сравнения digest — без Docker это невозможно. Заметим, что решение к воспроизводимости и не стремится: базовый образ взят по тегу python:3.13-slim, а не по digest. Для проекта 1 это осознанное упрощение; в проекте 4 так делать уже нельзя.

Зависимости не зафиксированы файлом блокировки. У приложения их нет вовсе — только стандартная библиотека. Как только появится первая, понадобится uv.lock или requirements.lock (урок 6.4).

Обработка сигналов не добавлена. Для инструмента, живущего доли секунды, SIGTERM не наступает: процесс завершается раньше. KeyboardInterrupt обработан, потому что он как раз случается — при Ctrl+C на большом файле. В проекте 2, где сервис живёт долго, обработка сигналов становится обязательной.

Сравнение с вашей реализацией

Стоит сравнить по пяти вопросам — они дают больше, чем чтение кода:

Что происходит с неразобранной строкой? Пропуск, отказ или молчание. Молчание — единственный неверный ответ.
Сколько у вас кодов возврата и различают ли они «пусто» и «ошибка»?
Можно ли по выводу понять, откуда взялось значение параметра?
Проверяют ли ваши тесты stdout и stderr раздельно?
Что произойдёт при | head?

Если по какому-то вопросу ваше решение обоснованнее — так и есть. Пятый вопрос, например, для инструмента, никогда не попадающего в конвейер, можно счесть несущественным.